package alice.datasegment;

import org.msgpack.type.Value;

import alice.codesegment.CodeSegment;
import alice.datasegment.CommandType;

public class LocalDataSegmentManager extends DataSegmentManager {
	
	public LocalDataSegmentManager() {
		new Thread(replyThread).start();
	}
	
	public DataSegmentKey getDataSegmentKey(String key) {
		if (key == null) {
			return null;
		}
		DataSegmentKey newDataSegmentKey = new DataSegmentKey();
		DataSegmentKey dataSegmentKey = dataSegments.putIfAbsent(key, newDataSegmentKey);
		if (dataSegmentKey == null) {
			newDataSegmentKey.runKeyThread();
			dataSegmentKey = newDataSegmentKey;
		}
		return dataSegmentKey;
	}
	
	@Override
	public void put(String key, Value val) {
		DataSegmentKey dataSegmentKey = getDataSegmentKey(key);
		dataSegmentKey.addCommand(new Command(CommandType.PUT, null, null, val, 0, 0, replyQueue, null));
	}

	@Override
	public void update(String key, Value val) {
		DataSegmentKey dataSegmentKey = getDataSegmentKey(key);
		dataSegmentKey.addCommand(new Command(CommandType.UPDATE, null, null, val, 0, 0, replyQueue, null));
	}

	@Override
	public void take(String argKey, String key, int index, CodeSegment cs) {
		DataSegmentKey dataSegmentKey = getDataSegmentKey(key);
		int seq = this.seq.getAndIncrement();
		Command cmd = new Command(CommandType.TAKE, argKey, null, null, index, seq, replyQueue, cs);
		seqHash.put(seq, cmd);
		dataSegmentKey.addCommand(cmd);
	}

	@Override
	public void peek(String argKey, String key, int index, CodeSegment cs) {
		DataSegmentKey dataSegmentKey = getDataSegmentKey(key);
		int seq = this.seq.getAndIncrement();
		Command cmd = new Command(CommandType.PEEK, argKey, null, null, index, seq, replyQueue, cs);
		seqHash.put(seq, cmd);
		dataSegmentKey.addCommand(cmd);
	}

	@Override
	public void remove(String key) {
		DataSegmentKey dataSegmentKey = getDataSegmentKey(key);
		dataSegmentKey.addCommand(new Command(CommandType.REMOVE, null, null, null, 0, 0, replyQueue, null));
	}

}