package alice.daemon;

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

import org.msgpack.unpacker.Unpacker;

import alice.codesegment.SingletonMessage;
import alice.datasegment.Command;
import alice.datasegment.CommandType;
import alice.datasegment.DataSegment;
import alice.datasegment.DataSegmentManager;
import alice.datasegment.LocalDataSegmentManager;
import alice.datasegment.ReceiveData;
import alice.topology.manager.keeparive.RespondData;

public class IncomingTcpConnection extends Thread {

    private Connection connection;
    protected DataSegmentManager manager;
    protected String reverseKey;
    private LocalDataSegmentManager lmanager = DataSegment.getLocal();

    public IncomingTcpConnection(DataSegmentManager manager) {
        this.manager = manager;
    }

    public IncomingTcpConnection(Connection connection, DataSegmentManager manager, String reverseKey) {
        this.manager = manager;
        this.connection = connection;
        this.reverseKey = reverseKey;
    }

    public LocalDataSegmentManager getLocalDataSegmentManager(){
        return lmanager;
    }

    /**
     * pipeline thread for receiving
     */
    public void run() {
        Unpacker unpacker = null;
        try {
            unpacker = SingletonMessage.getInstance().createUnpacker(connection.socket.getInputStream());
        } catch (IOException e) {
            e.printStackTrace();
        }
        if (unpacker == null) {
            return;
        }
        while (true) {
            try {
                Command cmd = null;
                ReceiveData rData = null;
                CommandMessage msg = unpacker.read(CommandMessage.class);
                CommandType type = CommandType.getCommandTypeFromId(msg.type);
                switch (type) {
                case UPDATE:
                case PUT:
                    rData = new ReceiveData(getSerializedByteArray(unpacker), msg.compressed, msg.serialized);
                    if (msg.setTime) {
                        rData.setTime = true;
                        rData.time = msg.time;
                        rData.depth = msg.depth;
                    }
                    cmd = new Command(type, null, null, rData, 0, 0, null, null, reverseKey);
                    lmanager.getDataSegmentKey(msg.key).runCommand(cmd);
                    break;
                case PEEK:
                case TAKE:
                    cmd = new Command(type, null, null, null, msg.index, msg.seq, null, null, connection);
                    cmd.setQuickFlag(msg.quickFlag);
                    lmanager.getDataSegmentKey(msg.key).runCommand(cmd);
                    break;
                case REMOVE:
                    cmd = new Command(type, null, null, null, 0, 0, null, null, "");
                    lmanager.getDataSegmentKey(msg.key).runCommand(cmd);
                    break;
                case REPLY:
                    cmd = manager.getAndRemoveCmd(msg.seq);
                    rData = new ReceiveData(getSerializedByteArray(unpacker), msg.compressed, msg.serialized);
                    Command rCmd = new Command(type, null, null, rData, msg.index, msg.seq, null, null, "");
                    cmd.cs.ids.reply(cmd.receiver, rCmd);
                    break;
                case PING:
                    if (DataSegment.contains(reverseKey))
                        DataSegment.get(reverseKey).response(msg.key);
                    break;
                case RESPONSE:
                    rData = new ReceiveData(new RespondData(reverseKey, System.currentTimeMillis()), false, false);
                    DataSegment.getLocal().put(msg.key, rData, null);
                    break;
                default:
                    break;
                }
            } catch (ClosedChannelException e) {
                return;
            } catch (EOFException e) {
                return;
            } catch (IOException e) {
                return;
            }
        }

    }

    private byte[] getSerializedByteArray(Unpacker unpacker) {
        int len;
        byte[] b = null;
        try {
            len = unpacker.readInt();
            b = unpacker.getSerializedByteArray(len);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return b;
    }

    public void setReverseKey(String name) {
        this.reverseKey = name;
        setName(name+"-IncomingTcp");
    }
}