package alice.datasegment;

import org.msgpack.type.Value;

import alice.codesegment.CodeSegment;
import alice.daemon.Connection;
import alice.daemon.IncomingTcpConnection;
import alice.daemon.OutboundTcpConnection;

public class RemoteDataSegmentManager extends DataSegmentManager {
	
	Connection connection;
	
	public RemoteDataSegmentManager(Connection connection) {
		this.connection = connection;
		new IncomingTcpConnection(connection, this).start();
		new OutboundTcpConnection(connection).start();
		new Thread(replyThread, "RemoteDataSegmentManager-"
		+ connection.socket.getInetAddress().getHostName()
		+ ":" + connection.socket.getPort()).start();
	}
	
	@Override
	public void put(String key, Value val) {
		connection.sendCommand(new Command(CommandType.PUT, null, key, val, 0, 0, null, null));
	}

	@Override
	public void update(String key, Value val) {
		connection.sendCommand(new Command(CommandType.UPDATE, null, key, val, 0, 0, null, null));
	}

	@Override
	public void take(DataSegmentReceiver receiver, String key, int index, CodeSegment cs) {
		int seq = this.seq.getAndIncrement();
		Command cmd = new Command(CommandType.TAKE, receiver, key, null, index, seq, replyQueue, cs);
		seqHash.put(seq, cmd);
		connection.sendCommand(cmd);		
	}

	@Override
	public void peek(DataSegmentReceiver receiver, String key, int index, CodeSegment cs) {
		int seq = this.seq.getAndIncrement();
		Command cmd = new Command(CommandType.PEEK, receiver, key, null, index, seq, replyQueue, cs);
		seqHash.put(seq, cmd);
		connection.sendCommand(cmd);
	}

	@Override
	public void remove(String key) {
		connection.sendCommand(new Command(CommandType.REMOVE, null, key, null, 0, 0, null, null));
	}

}