package com.alibaba.otter.canal.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Arrays;

import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;

import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
import com.alibaba.otter.canal.instance.manager.CanalInstanceWithManager;
import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.HAMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.IndexMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
import com.alibaba.otter.canal.protocol.CanalPacket.Ack;
import com.alibaba.otter.canal.protocol.CanalPacket.ClientAck;
import com.alibaba.otter.canal.protocol.CanalPacket.ClientAuth;
import com.alibaba.otter.canal.protocol.CanalPacket.ClientRollback;
import com.alibaba.otter.canal.protocol.CanalPacket.Get;
import com.alibaba.otter.canal.protocol.CanalPacket.Handshake;
import com.alibaba.otter.canal.protocol.CanalPacket.Messages;
import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
import com.alibaba.otter.canal.protocol.CanalPacket.Sub;
import com.alibaba.otter.canal.protocol.CanalPacket.Unsub;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
@Ignore
public class CanalServerTest {

    protected static final String cluster1      = "127.0.0.1:2188";
    protected static final String DESTINATION   = "ljhtest1";
    protected static final String DETECTING_SQL = "insert into retl.xdual values(1,now()) on duplicate key update x=now()";
    protected static final String MYSQL_ADDRESS = "127.0.0.1";
    protected static final String USERNAME      = "retl";
    protected static final String PASSWORD      = "retl";
    protected static final String FILTER        = "retl\\..*,erosa.canaltable1s,erosa.canaltable1t";

    private final ByteBuffer      header        = ByteBuffer.allocate(4);
    private CanalServerWithNetty  nettyServer;

    @Before
    public void setUp() {
        CanalServerWithEmbedded embeddedServer = new CanalServerWithEmbedded();
        embeddedServer.setCanalInstanceGenerator(new CanalInstanceGenerator() {

            public CanalInstance generate(String destination) {
                Canal canal = buildCanal();
                return new CanalInstanceWithManager(canal, FILTER);
            }
        });

        nettyServer = CanalServerWithNetty.instance();
        nettyServer.setEmbeddedServer(embeddedServer);
        nettyServer.setPort(1088);
        nettyServer.start();
    }

    @Test
    public void testAuth() {


        try {
            SocketChannel channel = SocketChannel.open();
            channel.connect(new InetSocketAddress("127.0.0.1", 1088));
            Packet p = Packet.parseFrom(readNextPacket(channel));

            if (p.getVersion() != 1) {
                throw new Exception("unsupported version at this client.");
            }

            if (p.getType() != PacketType.HANDSHAKE) {
                throw new Exception("expect handshake but found other type.");
            }
            //
            Handshake handshake = Handshake.parseFrom(p.getBody());
            System.out.println(handshake.getSupportedCompressions());
            //
            ClientAuth ca = ClientAuth.newBuilder()
                .setUsername("")
                .setNetReadTimeout(10000)
                .setNetWriteTimeout(10000)
                .build();
            writeWithHeader(channel,
                Packet.newBuilder()
                    .setType(PacketType.CLIENTAUTHENTICATION)
                    .setBody(ca.toByteString())
                    .build()
                    .toByteArray());
            //
            p = Packet.parseFrom(readNextPacket(channel));
            if (p.getType() != PacketType.ACK) {
                throw new Exception("unexpected packet type when ack is expected");
            }

            Ack ack = Ack.parseFrom(p.getBody());
            if (ack.getErrorCode() > 0) {
                throw new Exception("something goes wrong when doing authentication: " + ack.getErrorMessage());
            }

            writeWithHeader(channel,
                Packet.newBuilder()
                    .setType(PacketType.SUBSCRIPTION)
                    .setBody(Sub.newBuilder().setDestination(DESTINATION).setClientId("1").build().toByteString())
                    .build()
                    .toByteArray());
            //
            p = Packet.parseFrom(readNextPacket(channel));
            ack = Ack.parseFrom(p.getBody());
            if (ack.getErrorCode() > 0) {
                throw new Exception("failed to subscribe with reason: " + ack.getErrorMessage());
            }

            for (int i = 0; i < 10; i++) {
                writeWithHeader(channel,
                    Packet.newBuilder()
                        .setType(PacketType.GET)
                        .setBody(Get.newBuilder()
                            .setDestination(DESTINATION)
                            .setClientId("1")
                            .setFetchSize(10)
                            .build()
                            .toByteString())
                        .build()
                        .toByteArray());
                p = Packet.parseFrom(readNextPacket(channel));

                long batchId = -1L;
                switch (p.getType()) {
                    case MESSAGES: {
                        Messages messages = Messages.parseFrom(p.getBody());
                        batchId = messages.getBatchId();
                        break;
                    }
                    case ACK: {
                        ack = Ack.parseFrom(p.getBody());
                        if (ack.getErrorCode() > 0) {
                            throw new Exception("failed to subscribe with reason: " + ack.getErrorMessage());
                        }
                        break;
                    }
                    default: {
                        throw new Exception("unexpected packet type: " + p.getType());
                    }
                }

                System.out.println("!!!!!!!!!!!!!!!!! " + batchId);
                Thread.sleep(1000L);
                writeWithHeader(channel,
                    Packet.newBuilder()
                        .setType(PacketType.CLIENTACK)
                        .setBody(ClientAck.newBuilder()
                            .setDestination(DESTINATION)
                            .setClientId("1")
                            .setBatchId(batchId)
                            .build()
                            .toByteString())
                        .build()
                        .toByteArray());
            }

            writeWithHeader(channel,
                Packet.newBuilder()
                    .setType(PacketType.CLIENTROLLBACK)
                    .setBody(ClientRollback.newBuilder()
                        .setDestination(DESTINATION)
                        .setClientId("1")
                        .build()
                        .toByteString())
                    .build()
                    .toByteArray());

            writeWithHeader(channel,
                Packet.newBuilder()
                    .setType(PacketType.UNSUBSCRIPTION)
                    .setBody(Unsub.newBuilder().setDestination(DESTINATION).setClientId("1").build().toByteString())
                    .build()
                    .toByteArray());

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @After
    public void tearDown() {
        nettyServer.stop();
    }

    private byte[] readNextPacket(SocketChannel channel) throws IOException {
        header.clear();
        read(channel, header);
        int bodyLen = header.getInt(0);
        ByteBuffer bodyBuf = ByteBuffer.allocate(bodyLen);
        read(channel, bodyBuf);
        return bodyBuf.array();
    }

    private void writeWithHeader(SocketChannel channel, byte[] body) throws IOException {
        ByteBuffer header = ByteBuffer.allocate(4);
        header.putInt(body.length);
        header.flip();
        int len = channel.write(header);
        assert (len == header.capacity());

        channel.write(ByteBuffer.wrap(body));
    }

    private void read(SocketChannel channel, ByteBuffer buffer) throws IOException {
        while (buffer.hasRemaining()) {
            int r = channel.read(buffer);
            if (r == -1) {
                throw new IOException("end of stream when reading header");
            }
        }
    }

    private Canal buildCanal() {
        Canal canal = new Canal();
        canal.setId(1L);
        canal.setName(DESTINATION);
        canal.setDesc("test");

        CanalParameter parameter = new CanalParameter();

        parameter.setZkClusters(Arrays.asList("127.0.0.1:2188"));
        parameter.setMetaMode(MetaMode.MEMORY);
        parameter.setHaMode(HAMode.HEARTBEAT);
        parameter.setIndexMode(IndexMode.MEMORY);

        parameter.setStorageMode(StorageMode.MEMORY);
        parameter.setMemoryStorageBufferSize(32 * 1024);

        parameter.setSourcingType(SourcingType.MYSQL);
        parameter.setDbAddresses(Arrays.asList(new InetSocketAddress(MYSQL_ADDRESS, 3306),
            new InetSocketAddress(MYSQL_ADDRESS, 3306)));
        parameter.setDbUsername(USERNAME);
        parameter.setDbPassword(PASSWORD);
        parameter.setPositions(Arrays.asList("{\"journalName\":\"mysql-bin.000001\",\"position\":6163L,\"timestamp\":1322803601000L}",
            "{\"journalName\":\"mysql-bin.000001\",\"position\":6163L,\"timestamp\":1322803601000L}"));

        parameter.setSlaveId(1234L);

        parameter.setDefaultConnectionTimeoutInSeconds(30);
        parameter.setConnectionCharset("UTF-8");
        parameter.setConnectionCharsetNumber((byte) 33);
        parameter.setReceiveBufferSize(8 * 1024);
        parameter.setSendBufferSize(8 * 1024);

        parameter.setDetectingEnable(false);
        parameter.setDetectingIntervalInSeconds(10);
        parameter.setDetectingRetryTimes(3);
        parameter.setDetectingSQL(DETECTING_SQL);

        canal.setCanalParameter(parameter);
        return canal;
    }
}
