package alice.daemon;

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

import alice.datasegment.*;
import org.msgpack.MessagePack;
import org.msgpack.unpacker.Unpacker;

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();
    private CompressedLocalDataSegmentManager compressedlmanager = DataSegment.getCompressedLocal();
    private static final MessagePack packer = new MessagePack();

    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;
    }

    public CompressedLocalDataSegmentManager getCompressedLocalDataSegmentManager(){
        return compressedlmanager;
    }

    /**
     * pipeline thread for receiving
     */
    public void run() {
        Unpacker unpacker = null;
        try {
            unpacker = packer.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);///read header
                CommandType type = CommandType.getCommandTypeFromId(msg.type);
                switch (type) {
                case UPDATE:
                case PUT:
                    int dataSize = unpacker.readInt();
                    //byte[] data = unpacker.getSerializedByteArray(dataSize);
                    byte[] data = new byte[dataSize];
                    connection.socket.getInputStream().read(data);
                    rData = new ReceiveData(data, msg.compressed, msg.dataSize);

                    /*
                    rData = new ReceiveData(new byte[dataSize], msg.compressed, msg.dataSize);
                    try {
                        unpacker.readPayload(rData.getMessagePack());
                    } catch ( );
                      */


                    if (msg.setTime) {
                        rData.setTimes(msg.time, true, msg.depth);
                    }
                    if (msg.setZepped){
                        rData.setZipped(msg.zippedDataSize, true);
                    } else {
                        rData.setZipped(dataSize, true);
                    }

                    cmd = new Command(type, null, null, rData, 0, 0, null, null, reverseKey);
                    cmd.setCompressFlag(msg.compressed);

                    if (rData.compressed()){
                        compressedlmanager.getDataSegmentKey(msg.key).runCommand(cmd);
                    } else {
                        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.setCompressFlag(msg.compressed);

                    if (msg.compressed){
                        compressedlmanager.getDataSegmentKey(msg.key).runCommand(cmd);
                    } else {
                        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);//ToDo:fix
                    break;
                case REPLY:
                    cmd = manager.getAndRemoveCmd(msg.seq);

                    rData = new ReceiveData(unpacker.getSerializedByteArray(unpacker.readInt()), msg.compressed, msg.dataSize);

                    Command rCmd = new Command(type, null, null, rData, msg.index, msg.seq, null, null, "");
                    cmd.setCompressFlag(msg.compressed);
                    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()));
                    DataSegment.getLocal().put(msg.key, rData, false);
                    break;
                default:
                    break;
                }
            } catch (ClosedChannelException e) {
                return;
            } catch (EOFException e) {
                return;
            } catch (IOException e) {
                return;
            }
        }
    }

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