package alice.daemon;

import java.io.EOFException;
import java.io.IOException;
import java.nio.channels.ClosedChannelException;

import org.msgpack.MessagePack;
import org.msgpack.unpacker.Unpacker;

import alice.datasegment.Command;
import alice.datasegment.CommandType;
import alice.datasegment.DataSegment;
import alice.datasegment.DataSegmentKey;
import alice.datasegment.DataSegmentManager;
import alice.datasegment.LocalDataSegmentManager;

public class IncomingTcpConnection extends Thread {
	
	private static MessagePack MSGPACK = new MessagePack();
	public Connection connection;
	public DataSegmentManager manager;
	public String reverseKey;
	public IncomingTcpConnection(Connection connection, DataSegmentManager manager, String reverseKey) {
		this.manager = manager;
		this.connection = connection;
		this.reverseKey = reverseKey;
	}
	
	/**
	 * pipeline thread for receiving
	 */
	public void run() {
		Unpacker unpacker = null;
		try {
			unpacker = MSGPACK.createUnpacker(connection.socket.getInputStream());
		} catch (IOException e2) {
			e2.printStackTrace();
		}
		while (true) {
			try {
				CommandMessage msg = unpacker.read(CommandMessage.class);
				CommandType type = CommandType.getCommandTypeFromId(msg.type);
				switch (type) {
				case UPDATE:
					getDataSegmentKey(msg).addCommand(new Command(type, null, null, msg.val, 0, 0, null, null, reverseKey));
					break;
				case PUT:
					getDataSegmentKey(msg).addCommand(new Command(type, null, null, msg.val, 0, 0, null, null, reverseKey));
					break;
				case PEEK:
					//Command(CommandType cmdType, String argKey, Value val, int index, int seq, BlockingQueue<Command> replyQueue, CodeSegment cs) {
					getDataSegmentKey(msg).addCommand(new Command(type, null, null, null, msg.index, msg.seq, connection.sendQueue, null, null));
					break;
				case TAKE:
					getDataSegmentKey(msg).addCommand(new Command(type, null, null, null, msg.index, msg.seq, connection.sendQueue, null, null));
					break;	
				case REMOVE:
					getDataSegmentKey(msg).addCommand(new Command(type, null, null, null, 0, 0, null, null, null));
					break;
				case REPLY:
					try {
						manager.replyQueue.put(new Command(type, null, null, msg.val, msg.index, msg.seq, null, null, null));
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					break;
				default:
					break;
				}
			} catch (ClosedChannelException e) {
				try {
					connection.sendQueue.put(new Command(CommandType.CLOSE, null, null, null, 0, 0, null, null, null));
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}
				return;
			} catch (EOFException e) {
				try {
					connection.sendQueue.put(new Command(CommandType.CLOSE, null, null, null, 0, 0, null, null, null));
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}
				return;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	private DataSegmentKey getDataSegmentKey(CommandMessage msg) {
		LocalDataSegmentManager lmanager = DataSegment.getLocal();
		return lmanager.getDataSegmentKey(msg.key);
	}
}