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:
                    System.out.println("in TCP PUT");
                    rData = new ReceiveData(packer.read(unpacker.getSerializedByteArray(unpacker.readInt()), byte[].class), msg.compressed, msg.dataSize);///read rData

                    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);
                    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:
                    System.out.println("in TCP 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:
                    System.out.println("in TCP REPLY");
                    cmd = manager.getAndRemoveCmd(msg.seq);
                    // ToDo: do not pack again on compressedMessagePack
                    // after size of copressedData get to ByteBuffer of compressed data using
                    // readPayloadAsReference().toByteBuffer()
                    rData = new ReceiveData(packer.read(unpacker.getSerializedByteArray(unpacker.readInt()), byte[].class), 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, null);
                    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");
    }
}