cannal源码赏析 - server模块

public class CanalServerWithNetty extends AbstractCanalLifeCycle implements CanalServer {

    private CanalServerWithEmbedded embeddedServer;      // 嵌入式server
    private String                  ip;
    private int                     port;
    private Channel                 serverChannel = null;
    private ServerBootstrap         bootstrap     = null;
    private ChannelGroup            childGroups   = null; // socket channel
                                                          // container, used to
                                                          // close sockets
                                                          // explicitly.

    private static class SingletonHolder {

        private static final CanalServerWithNetty CANAL_SERVER_WITH_NETTY = new CanalServerWithNetty();
    }

    private CanalServerWithNetty(){
        this.embeddedServer = CanalServerWithEmbedded.instance();
        this.childGroups = new DefaultChannelGroup();
    }

    public static CanalServerWithNetty instance() {
        return SingletonHolder.CANAL_SERVER_WITH_NETTY;
    }

    public void start() {
        super.start();

        if (!embeddedServer.isStart()) {
            embeddedServer.start();
        }

        this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
            Executors.newCachedThreadPool()));
        /*
         * enable keep-alive mechanism, handle abnormal network connection
         * scenarios on OS level. the threshold parameters are depended on OS.
         * e.g. On Linux: net.ipv4.tcp_keepalive_time = 300
         * net.ipv4.tcp_keepalive_probes = 2 net.ipv4.tcp_keepalive_intvl = 30
         */
        bootstrap.setOption("child.keepAlive", true);
        /*
         * optional parameter.
         */
        bootstrap.setOption("child.tcpNoDelay", true);

        // 构造对应的pipeline
        bootstrap.setPipelineFactory(() -> {
            ChannelPipeline pipelines = Channels.pipeline();
            pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());
            // support to maintain child socket channel.
            pipelines.addLast(HandshakeInitializationHandler.class.getName(),
                new HandshakeInitializationHandler(childGroups));
            pipelines.addLast(ClientAuthenticationHandler.class.getName(),
                new ClientAuthenticationHandler(embeddedServer));

            SessionHandler sessionHandler = new SessionHandler(embeddedServer);
            pipelines.addLast(SessionHandler.class.getName(), sessionHandler);
            return pipelines;
        });

        // 启动
        if (StringUtils.isNotEmpty(ip)) {
            this.serverChannel = bootstrap.bind(new InetSocketAddress(this.ip, this.port));
        } else {
            this.serverChannel = bootstrap.bind(new InetSocketAddress(this.port));
        }
    }

    public void stop() {
        super.stop();

        if (this.serverChannel != null) {
            this.serverChannel.close().awaitUninterruptibly(1000);
        }

        // close sockets explicitly to reduce socket channel hung in complicated
        // network environment.
        if (this.childGroups != null) {
            this.childGroups.close().awaitUninterruptibly(5000);
        }

        if (this.bootstrap != null) {
            this.bootstrap.releaseExternalResources();
        }

        if (embeddedServer.isStart()) {
            embeddedServer.stop();
        }
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
        this.embeddedServer = embeddedServer;
    }

}

public class FixedHeaderFrameDecoder extends ReplayingDecoder<VoidEnum> {

    protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, VoidEnum state)
                                                                                                             throws Exception {
        return buffer.readBytes(buffer.readInt());
    }
}

public class ClientAuthenticationHandler extends SimpleChannelHandler {

    private static final Logger     logger                                  = LoggerFactory.getLogger(ClientAuthenticationHandler.class);
    private final int               SUPPORTED_VERSION                       = 3;
    private final int               defaultSubscriptorDisconnectIdleTimeout = 60 * 60 * 1000;
    private CanalServerWithEmbedded embeddedServer;
    private byte[]                  seed;

    public ClientAuthenticationHandler(){

    }

    public ClientAuthenticationHandler(CanalServerWithEmbedded embeddedServer){
        this.embeddedServer = embeddedServer;
    }

    public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
        final Packet packet = Packet.parseFrom(buffer.readBytes(buffer.readableBytes()).array());
        switch (packet.getVersion()) {
            case SUPPORTED_VERSION:
            default:
                final ClientAuth clientAuth = ClientAuth.parseFrom(packet.getBody());
                if (seed == null) {
                    byte[] errorBytes = NettyUtils.errorPacket(400,
                        MessageFormatter.format("auth failed for seed is null", clientAuth.getUsername()).getMessage());
                    NettyUtils.write(ctx.getChannel(), errorBytes, null);
                    break;
                }

                if (!embeddedServer.auth(clientAuth.getUsername(), clientAuth.getPassword().toStringUtf8(), seed)) {
                    byte[] errorBytes = NettyUtils.errorPacket(400,
                        MessageFormatter.format("auth failed for user:{}", clientAuth.getUsername()).getMessage());
                    NettyUtils.write(ctx.getChannel(), errorBytes, null);
                    break;
                }

                // 如果存在订阅信息
                if (StringUtils.isNotEmpty(clientAuth.getDestination())
                    && StringUtils.isNotEmpty(clientAuth.getClientId())) {
                    ClientIdentity clientIdentity = new ClientIdentity(clientAuth.getDestination(),
                        Short.valueOf(clientAuth.getClientId()),
                        clientAuth.getFilter());
                    try {
                        MDC.put("destination", clientIdentity.getDestination());
                        embeddedServer.subscribe(clientIdentity);
                        // 尝试启动，如果已经启动，忽略
                        if (!embeddedServer.isStart(clientIdentity.getDestination())) {
                            ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
                            if (!runningMonitor.isStart()) {
                                runningMonitor.start();
                            }
                        }
                    } finally {
                        MDC.remove("destination");
                    }
                }
                // 鉴权一次性，暂不统计
                NettyUtils.ack(ctx.getChannel(), future -> {
                    logger.info("remove unused channel handlers after authentication is done successfully.");
                    ctx.getPipeline().remove(HandshakeInitializationHandler.class.getName());
                    ctx.getPipeline().remove(ClientAuthenticationHandler.class.getName());

                    int readTimeout = defaultSubscriptorDisconnectIdleTimeout;
                    int writeTimeout = defaultSubscriptorDisconnectIdleTimeout;
                    if (clientAuth.getNetReadTimeout() > 0) {
                        readTimeout = clientAuth.getNetReadTimeout();
                    }
                    if (clientAuth.getNetWriteTimeout() > 0) {
                        writeTimeout = clientAuth.getNetWriteTimeout();
                    }
                    // fix bug: soTimeout parameter's unit from connector is
                    // millseconds.
                    IdleStateHandler idleStateHandler = new IdleStateHandler(NettyUtils.hashedWheelTimer,
                        readTimeout,
                        writeTimeout,
                        0,
                        TimeUnit.MILLISECONDS);
                    ctx.getPipeline().addBefore(SessionHandler.class.getName(),
                        IdleStateHandler.class.getName(),
                        idleStateHandler);

                    IdleStateAwareChannelHandler idleStateAwareChannelHandler = new IdleStateAwareChannelHandler() {

                        public void channelIdle(ChannelHandlerContext ctx1, IdleStateEvent e1) throws Exception {
                            logger.warn("channel:{} idle timeout exceeds, close channel to save server resources...",
                                ctx1.getChannel());
                            ctx1.getChannel().close();
                        }

                    };
                    ctx.getPipeline().addBefore(SessionHandler.class.getName(),
                        IdleStateAwareChannelHandler.class.getName(),
                        idleStateAwareChannelHandler);
                });
                break;
        }
    }

    public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
        this.embeddedServer = embeddedServer;
    }

    public void setSeed(byte[] seed) {
        this.seed = seed;
    }

}

public class HandshakeInitializationHandler extends SimpleChannelHandler {

    // support to maintain socket channel.
    private ChannelGroup childGroups;

    public HandshakeInitializationHandler(ChannelGroup childGroups){
        this.childGroups = childGroups;
    }

    private static final Logger logger = LoggerFactory.getLogger(HandshakeInitializationHandler.class);

    public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        // add new socket channel in channel container, used to manage sockets.
        if (childGroups != null) {
            childGroups.add(ctx.getChannel());
        }

        final byte[] seed = org.apache.commons.lang3.RandomUtils.nextBytes(8);
        byte[] body = Packet.newBuilder()
            .setType(CanalPacket.PacketType.HANDSHAKE)
            .setVersion(NettyUtils.VERSION)
            .setBody(Handshake.newBuilder().setSeeds(ByteString.copyFrom(seed)).build().toByteString())
            .build()
            .toByteArray();

        NettyUtils.write(ctx.getChannel(), body, future -> {
            ctx.getPipeline().get(HandshakeInitializationHandler.class.getName());
            ClientAuthenticationHandler handler = (ClientAuthenticationHandler) ctx.getPipeline()
                .get(ClientAuthenticationHandler.class.getName());
            handler.setSeed(seed);
        });
        logger.info("send handshake initialization packet to : {}", ctx.getChannel());
    }
}

public class SessionHandler extends SimpleChannelHandler {

    private static final Logger     logger = LoggerFactory.getLogger(SessionHandler.class);
    private CanalServerWithEmbedded embeddedServer;

    public SessionHandler(){
    }

    public SessionHandler(CanalServerWithEmbedded embeddedServer){
        this.embeddedServer = embeddedServer;
    }

    @SuppressWarnings({ "deprecation" })
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        logger.info("message receives in session handler...");
        long start = System.nanoTime();
        ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
        Packet packet = Packet.parseFrom(buffer.readBytes(buffer.readableBytes()).array());
        ClientIdentity clientIdentity = null;
        try {
            switch (packet.getType()) {
                case SUBSCRIPTION:
                    Sub sub = Sub.parseFrom(packet.getBody());
                    if (StringUtils.isNotEmpty(sub.getDestination()) && StringUtils.isNotEmpty(sub.getClientId())) {
                        clientIdentity = new ClientIdentity(sub.getDestination(),
                            Short.valueOf(sub.getClientId()),
                            sub.getFilter());
                        MDC.put("destination", clientIdentity.getDestination());

                        // 尝试启动，如果已经启动，忽略
                        if (!embeddedServer.isStart(clientIdentity.getDestination())) {
                            ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
                            if (!runningMonitor.isStart()) {
                                runningMonitor.start();
                            }
                        }

                        embeddedServer.subscribe(clientIdentity);
                        // ctx.setAttachment(clientIdentity);// 设置状态数据
                        byte[] ackBytes = NettyUtils.ackPacket();
                        NettyUtils.write(ctx.getChannel(), ackBytes, new ChannelFutureAggregator(sub.getDestination(),
                            sub,
                            packet.getType(),
                            ackBytes.length,
                            System.nanoTime() - start));
                    } else {
                        byte[] errorBytes = NettyUtils.errorPacket(401,
                            MessageFormatter.format("destination or clientId is null", sub.toString()).getMessage());
                        NettyUtils.write(ctx.getChannel(),
                            errorBytes,
                            new ChannelFutureAggregator(sub.getDestination(),
                                sub,
                                packet.getType(),
                                errorBytes.length,
                                System.nanoTime() - start,
                                (short) 401));
                    }
                    break;
                case UNSUBSCRIPTION:
                    Unsub unsub = Unsub.parseFrom(packet.getBody());
                    if (StringUtils.isNotEmpty(unsub.getDestination()) && StringUtils.isNotEmpty(unsub.getClientId())) {
                        clientIdentity = new ClientIdentity(unsub.getDestination(),
                            Short.valueOf(unsub.getClientId()),
                            unsub.getFilter());
                        MDC.put("destination", clientIdentity.getDestination());
                        embeddedServer.unsubscribe(clientIdentity);
                        stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭
                        byte[] ackBytes = NettyUtils.ackPacket();
                        NettyUtils.write(ctx.getChannel(),
                            ackBytes,
                            new ChannelFutureAggregator(unsub.getDestination(),
                                unsub,
                                packet.getType(),
                                ackBytes.length,
                                System.nanoTime() - start));
                    } else {
                        byte[] errorBytes = NettyUtils.errorPacket(401,
                            MessageFormatter.format("destination or clientId is null", unsub.toString()).getMessage());
                        NettyUtils.write(ctx.getChannel(),
                            errorBytes,
                            new ChannelFutureAggregator(unsub.getDestination(),
                                unsub,
                                packet.getType(),
                                errorBytes.length,
                                System.nanoTime() - start,
                                (short) 401));
                    }
                    break;
                case GET:
                    Get get = CanalPacket.Get.parseFrom(packet.getBody());
                    if (StringUtils.isNotEmpty(get.getDestination()) && StringUtils.isNotEmpty(get.getClientId())) {
                        clientIdentity = new ClientIdentity(get.getDestination(), Short.valueOf(get.getClientId()));
                        MDC.put("destination", clientIdentity.getDestination());
                        Message message = null;

                        // if (get.getAutoAck()) {
                        // if (get.getTimeout() == -1) {//是否是初始值
                        // message = embeddedServer.get(clientIdentity,
                        // get.getFetchSize());
                        // } else {
                        // TimeUnit unit = convertTimeUnit(get.getUnit());
                        // message = embeddedServer.get(clientIdentity,
                        // get.getFetchSize(), get.getTimeout(), unit);
                        // }
                        // } else {
                        if (get.getTimeout() == -1) {// 是否是初始值
                            message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());
                        } else {
                            TimeUnit unit = convertTimeUnit(get.getUnit());
                            message = embeddedServer.getWithoutAck(clientIdentity,
                                get.getFetchSize(),
                                get.getTimeout(),
                                unit);
                        }
                        // }

                        if (message.getId() != -1 && message.isRaw()) {
                            List<ByteString> rowEntries = message.getRawEntries();
                            // message size
                            int messageSize = 0;
                            messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, message.getId());

                            int dataSize = 0;
                            for (ByteString rowEntry : rowEntries) {
                                dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntry);
                            }
                            messageSize += dataSize;
                            messageSize += 1 * rowEntries.size();
                            // packet size
                            int size = 0;
                            size += com.google.protobuf.CodedOutputStream.computeEnumSize(3,
                                PacketType.MESSAGES.getNumber());
                            size += com.google.protobuf.CodedOutputStream.computeTagSize(5)
                                    + com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize)
                                    + messageSize;
                            // recyle bytes
                            // ByteBuffer byteBuffer = (ByteBuffer)
                            // ctx.getAttachment();
                            // if (byteBuffer != null && size <=
                            // byteBuffer.capacity()) {
                            // byteBuffer.clear();
                            // } else {
                            // byteBuffer =
                            // ByteBuffer.allocate(size).order(ByteOrder.BIG_ENDIAN);
                            // ctx.setAttachment(byteBuffer);
                            // }
                            // CodedOutputStream output =
                            // CodedOutputStream.newInstance(byteBuffer);
                            byte[] body = new byte[size];
                            CodedOutputStream output = CodedOutputStream.newInstance(body);
                            output.writeEnum(3, PacketType.MESSAGES.getNumber());

                            output.writeTag(5, WireFormat.WIRETYPE_LENGTH_DELIMITED);
                            output.writeRawVarint32(messageSize);
                            // message
                            output.writeInt64(1, message.getId());
                            for (ByteString rowEntry : rowEntries) {
                                output.writeBytes(2, rowEntry);
                            }
                            output.checkNoSpaceLeft();
                            NettyUtils.write(ctx.getChannel(), body, new ChannelFutureAggregator(get.getDestination(),
                                get,
                                packet.getType(),
                                body.length,
                                System.nanoTime() - start,
                                message.getId() == -1));

                            // output.flush();
                            // byteBuffer.flip();
                            // NettyUtils.write(ctx.getChannel(), byteBuffer,
                            // null);
                        } else {
                            Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
                            packetBuilder.setType(PacketType.MESSAGES).setVersion(NettyUtils.VERSION);

                            Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder();
                            messageBuilder.setBatchId(message.getId());
                            if (message.getId() != -1) {
                                if (message.isRaw() && !CollectionUtils.isEmpty(message.getRawEntries())) {
                                    messageBuilder.addAllMessages(message.getRawEntries());
                                } else if (!CollectionUtils.isEmpty(message.getEntries())) {
                                    for (Entry entry : message.getEntries()) {
                                        messageBuilder.addMessages(entry.toByteString());
                                    }
                                }
                            }
                            byte[] body = packetBuilder.setBody(messageBuilder.build().toByteString())
                                .build()
                                .toByteArray();
                            NettyUtils.write(ctx.getChannel(), body, new ChannelFutureAggregator(get.getDestination(),
                                get,
                                packet.getType(),
                                body.length,
                                System.nanoTime() - start,
                                message.getId() == -1));// 输出数据
                        }
                    } else {
                        byte[] errorBytes = NettyUtils.errorPacket(401,
                            MessageFormatter.format("destination or clientId is null", get.toString()).getMessage());
                        NettyUtils.write(ctx.getChannel(),
                            errorBytes,
                            new ChannelFutureAggregator(get.getDestination(),
                                get,
                                packet.getType(),
                                errorBytes.length,
                                System.nanoTime() - start,
                                (short) 401));
                    }
                    break;
                case CLIENTACK:
                    ClientAck ack = CanalPacket.ClientAck.parseFrom(packet.getBody());
                    MDC.put("destination", ack.getDestination());
                    if (StringUtils.isNotEmpty(ack.getDestination()) && StringUtils.isNotEmpty(ack.getClientId())) {
                        if (ack.getBatchId() == 0L) {
                            byte[] errorBytes = NettyUtils.errorPacket(402,
                                MessageFormatter.format("batchId should assign value", ack.toString()).getMessage());
                            NettyUtils.write(ctx.getChannel(),
                                errorBytes,
                                new ChannelFutureAggregator(ack.getDestination(),
                                    ack,
                                    packet.getType(),
                                    errorBytes.length,
                                    System.nanoTime() - start,
                                    (short) 402));
                        } else if (ack.getBatchId() == -1L) { // -1代表上一次get没有数据，直接忽略之
                            // donothing
                        } else {
                            clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId()));
                            embeddedServer.ack(clientIdentity, ack.getBatchId());
                            new ChannelFutureAggregator(ack.getDestination(),
                                ack,
                                packet.getType(),
                                0,
                                System.nanoTime() - start).operationComplete(null);
                        }
                    } else {
                        byte[] errorBytes = NettyUtils.errorPacket(401,
                            MessageFormatter.format("destination or clientId is null", ack.toString()).getMessage());
                        NettyUtils.write(ctx.getChannel(),
                            errorBytes,
                            new ChannelFutureAggregator(ack.getDestination(),
                                ack,
                                packet.getType(),
                                errorBytes.length,
                                System.nanoTime() - start,
                                (short) 401));
                    }
                    break;
                case CLIENTROLLBACK:
                    ClientRollback rollback = CanalPacket.ClientRollback.parseFrom(packet.getBody());
                    MDC.put("destination", rollback.getDestination());
                    if (StringUtils.isNotEmpty(rollback.getDestination())
                        && StringUtils.isNotEmpty(rollback.getClientId())) {
                        clientIdentity = new ClientIdentity(rollback.getDestination(),
                            Short.valueOf(rollback.getClientId()));
                        if (rollback.getBatchId() == 0L) {
                            embeddedServer.rollback(clientIdentity);// 回滚所有批次
                        } else {
                            embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次
                        }
                        new ChannelFutureAggregator(rollback.getDestination(),
                            rollback,
                            packet.getType(),
                            0,
                            System.nanoTime() - start).operationComplete(null);
                    } else {
                        byte[] errorBytes = NettyUtils.errorPacket(401,
                            MessageFormatter.format("destination or clientId is null", rollback.toString())
                                .getMessage());
                        NettyUtils.write(ctx.getChannel(),
                            errorBytes,
                            new ChannelFutureAggregator(rollback.getDestination(),
                                rollback,
                                packet.getType(),
                                errorBytes.length,
                                System.nanoTime() - start,
                                (short) 401));
                    }
                    break;
                default:
                    byte[] errorBytes = NettyUtils.errorPacket(400,
                        MessageFormatter.format("packet type={} is NOT supported!", packet.getType()).getMessage());
                    NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ctx.getChannel()
                        .getRemoteAddress()
                        .toString(), null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
                    break;
            }
        } catch (Throwable exception) {
            byte[] errorBytes = NettyUtils.errorPacket(400,
                MessageFormatter.format("something goes wrong with channel:{}, exception={}",
                    ctx.getChannel(),
                    ExceptionUtils.getStackTrace(exception)).getMessage());
            NettyUtils.write(ctx.getChannel(), errorBytes, new ChannelFutureAggregator(ctx.getChannel()
                .getRemoteAddress()
                .toString(), null, packet.getType(), errorBytes.length, System.nanoTime() - start, (short) 400));
        } finally {
            MDC.remove("destination");
        }
    }

    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
        logger.error("something goes wrong with channel:{}, exception={}",
            ctx.getChannel(),
            ExceptionUtils.getStackTrace(e.getCause()));

        ctx.getChannel().close();
    }

    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        // logger.info("remove binding subscription value object if any...");
        // ClientIdentity clientIdentity = (ClientIdentity) ctx.getAttachment();
        // // 如果唯一的订阅者都取消了订阅，直接关闭服务，针对内部版本模式下可以减少资源浪费
        // if (clientIdentity != null) {
        // stopCanalInstanceIfNecessary(clientIdentity);
        // }
    }

    private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) {
        List<ClientIdentity> clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination());
        if (clientIdentitys != null && clientIdentitys.size() == 1 && clientIdentitys.contains(clientIdentity)) {
            ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
            if (runningMonitor.isStart()) {
                runningMonitor.release();
            }
        }
    }

    private TimeUnit convertTimeUnit(int unit) {
        switch (unit) {
            case 0:
                return TimeUnit.NANOSECONDS;
            case 1:
                return TimeUnit.MICROSECONDS;
            case 2:
                return TimeUnit.MILLISECONDS;
            case 3:
                return TimeUnit.SECONDS;
            case 4:
                return TimeUnit.MINUTES;
            case 5:
                return TimeUnit.HOURS;
            case 6:
                return TimeUnit.DAYS;
            default:
                return TimeUnit.MILLISECONDS;
        }
    }

    public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
        this.embeddedServer = embeddedServer;
    }

}

public class ChannelFutureAggregator implements ChannelFutureListener {

    private ClientRequestResult result;

    public ChannelFutureAggregator(String destination, GeneratedMessageV3 request, CanalPacket.PacketType type, int amount, long latency, boolean empty) {
        this(destination, request, type, amount, latency, empty, (short) 0);
    }

    public ChannelFutureAggregator(String destination, GeneratedMessageV3 request, CanalPacket.PacketType type, int amount, long latency) {
        this(destination, request, type, amount, latency, false, (short) 0);
    }

    public ChannelFutureAggregator(String destination, GeneratedMessageV3 request, CanalPacket.PacketType type, int amount, long latency, short errorCode) {
        this(destination, request, type, amount, latency, false, errorCode);
    }

    private ChannelFutureAggregator(String destination, GeneratedMessageV3 request, CanalPacket.PacketType type, int amount, long latency, boolean empty, short errorCode) {
        this.result = new ClientRequestResult.Builder()
                .destination(destination)
                .type(type)
                .request(request)
                .amount(amount + HEADER_LENGTH)
                .latency(latency)
                .errorCode(errorCode)
                .empty(empty)
                .build();
    }

    @Override
    public void operationComplete(ChannelFuture future) {
        // profiling after I/O operation
        if (future != null && future.getCause() != null) {
            result.channelError = future.getCause();
        }
        profiler().profiling(result);
    }

    /**
     * Client request result pojo
     */
    public static class ClientRequestResult {

        private String                 destination;
        private CanalPacket.PacketType type;
        private GeneratedMessageV3       request;
        private int                    amount;
        private long                   latency;
        private short                  errorCode;
        private boolean                empty;
        private Throwable              channelError;

        private ClientRequestResult() {}

        private ClientRequestResult(Builder builder) {
            this.destination = Preconditions.checkNotNull(builder.destination);
            this.type = Preconditions.checkNotNull(builder.type);
            this.request = builder.request;
            this.amount = builder.amount;
            this.latency = builder.latency;
            this.errorCode = builder.errorCode;
            this.empty = builder.empty;
            this.channelError = builder.channelError;
        }

        // auto-generated
        public static class Builder {

            private String                 destination;
            private CanalPacket.PacketType type;
            private GeneratedMessageV3       request;
            private int                    amount;
            private long                   latency;
            private short                  errorCode;
            private boolean                empty;
            private Throwable              channelError;

            Builder destination(String destination) {
                this.destination = destination;
                return this;
            }

            Builder type(CanalPacket.PacketType type) {
                this.type = type;
                return this;
            }

            Builder request(GeneratedMessageV3 request) {
                this.request = request;
                return this;
            }

            Builder amount(int amount) {
                this.amount = amount;
                return this;
            }

            Builder latency(long latency) {
                this.latency = latency;
                return this;
            }

            Builder errorCode(short errorCode) {
                this.errorCode = errorCode;
                return this;
            }

            Builder empty(boolean empty) {
                this.empty = empty;
                return this;
            }

            public Builder channelError(Throwable channelError) {
                this.channelError = channelError;
                return this;
            }

            public Builder fromPrototype(ClientRequestResult prototype) {
                destination = prototype.destination;
                type = prototype.type;
                request = prototype.request;
                amount = prototype.amount;
                latency = prototype.latency;
                errorCode = prototype.errorCode;
                empty = prototype.empty;
                channelError = prototype.channelError;
                return this;
            }

            ClientRequestResult build() {
                return new ClientRequestResult(this);
            }
        }
        // getters
        public String getDestination() {
            return destination;
        }

        public CanalPacket.PacketType getType() {
            return type;
        }

        public GeneratedMessageV3 getRequest() {
            return request;
        }

        public int getAmount() {
            return amount;
        }

        public long getLatency() {
            return latency;
        }

        public short getErrorCode() {
            return errorCode;
        }

        public boolean getEmpty() {
            return empty;
        }

        public Throwable getChannelError() {
            return channelError;
        }
    }
}

/**
 * 嵌入式版本实现
 *
 * @author jianghang 2012-7-12 下午01:34:00
 * @author zebin.xuzb
 * @version 1.0.0
 */
public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, CanalService {

    private static final Logger        logger  = LoggerFactory.getLogger(CanalServerWithEmbedded.class);
    private Map<String, CanalInstance> canalInstances;
    // private Map<ClientIdentity, Position> lastRollbackPostions;
    private CanalInstanceGenerator     canalInstanceGenerator;
    private int                        metricsPort;
    private CanalMetricsService        metrics = NopCanalMetricsService.NOP;
    private String                     user;
    private String                     passwd;

    private static class SingletonHolder {

        private static final CanalServerWithEmbedded CANAL_SERVER_WITH_EMBEDDED = new CanalServerWithEmbedded();
    }

    public CanalServerWithEmbedded(){
        // 希望也保留用户new单独实例的需求,兼容历史
    }

    public static CanalServerWithEmbedded instance() {
        return SingletonHolder.CANAL_SERVER_WITH_EMBEDDED;
    }

    public void start() {
        if (!isStart()) {
            super.start();
            // 如果存在provider,则启动metrics service
            if(metricsPort > 0) {
                loadCanalMetrics();
                metrics.setServerPort(metricsPort);
                metrics.initialize();
            }
            canalInstances = MigrateMap.makeComputingMap(destination -> canalInstanceGenerator.generate(destination));
            // lastRollbackPostions = new MapMaker().makeMap();
        }
    }

    public void stop() {
        super.stop();
        for (Map.Entry<String, CanalInstance> entry : canalInstances.entrySet()) {
            try {
                CanalInstance instance = entry.getValue();
                if (instance.isStart()) {
                    try {
                        String destination = entry.getKey();
                        MDC.put("destination", destination);
                        entry.getValue().stop();
                        logger.info("stop CanalInstances[{}] successfully", destination);
                    } finally {
                        MDC.remove("destination");
                    }
                }
            } catch (Exception e) {
                logger.error(String.format("stop CanalInstance[%s] has an error", entry.getKey()), e);
            }
        }
        metrics.terminate();
    }

    public boolean auth(String user, String passwd, byte[] seed) {
        // 如果user/passwd密码为空,则任何用户账户都能登录
        if ((StringUtils.isEmpty(this.user) || StringUtils.equals(this.user, user))) {
            if (StringUtils.isEmpty(this.passwd)) {
                return true;
            } else if (StringUtils.isEmpty(passwd)) {
                // 如果server密码有配置,客户端密码为空,则拒绝
                return false;
            }

            try {
                byte[] passForClient = SecurityUtil.hexStr2Bytes(passwd);
                return SecurityUtil.scrambleServerAuth(passForClient, SecurityUtil.hexStr2Bytes(this.passwd), seed);
            } catch (NoSuchAlgorithmException e) {
                return false;
            }
        }

        return false;
    }

    public void start(final String destination) {
        final CanalInstance canalInstance = canalInstances.get(destination);
        if (!canalInstance.isStart()) {
            try {
                MDC.put("destination", destination);
                if (metrics.isRunning()) {
                    metrics.register(canalInstance);
                }
                canalInstance.start();
                logger.info("start CanalInstances[{}] successfully", destination);
            } finally {
                MDC.remove("destination");
            }
        }
    }

    public void stop(String destination) {
        CanalInstance canalInstance = canalInstances.remove(destination);
        if (canalInstance != null) {
            if (canalInstance.isStart()) {
                try {
                    MDC.put("destination", destination);
                    canalInstance.stop();
                    if (metrics.isRunning()) {
                        metrics.unregister(canalInstance);
                    }
                    logger.info("stop CanalInstances[{}] successfully", destination);
                } finally {
                    MDC.remove("destination");
                }
            }
        }
    }

    public boolean isStart(String destination) {
        return canalInstances.containsKey(destination) && canalInstances.get(destination).isStart();
    }

    /**
     * 客户端订阅，重复订阅时会更新对应的filter信息
     */
    @Override
    public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
        checkStart(clientIdentity.getDestination());

        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        if (!canalInstance.getMetaManager().isStart()) {
            canalInstance.getMetaManager().start();
        }

        canalInstance.getMetaManager().subscribe(clientIdentity); // 执行一下meta订阅

        Position position = canalInstance.getMetaManager().getCursor(clientIdentity);
        if (position == null) {
            position = canalInstance.getEventStore().getFirstPosition();// 获取一下store中的第一条
            if (position != null) {
                canalInstance.getMetaManager().updateCursor(clientIdentity, position); // 更新一下cursor
            }
            logger.info("subscribe successfully, {} with first position:{} ", clientIdentity, position);
        } else {
            logger.info("subscribe successfully, {} use last cursor position:{} ", clientIdentity, position);
        }

        // 通知下订阅关系变化
        canalInstance.subscribeChange(clientIdentity);
    }

    /**
     * 取消订阅
     */
    @Override
    public void unsubscribe(ClientIdentity clientIdentity) throws CanalServerException {
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        canalInstance.getMetaManager().unsubscribe(clientIdentity); // 执行一下meta订阅

        logger.info("unsubscribe successfully, {}", clientIdentity);
    }

    /**
     * 查询所有的订阅信息
     */
    public List<ClientIdentity> listAllSubscribe(String destination) throws CanalServerException {
        CanalInstance canalInstance = canalInstances.get(destination);
        return canalInstance.getMetaManager().listAllSubscribeInfo(destination);
    }

    /**
     * 获取数据
     *
     * <pre>
     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
     * </pre>
     */
    @Override
    public Message get(ClientIdentity clientIdentity, int batchSize) throws CanalServerException {
        return get(clientIdentity, batchSize, null, null);
    }

    /**
     * 获取数据，可以指定超时时间.
     *
     * <pre>
     * 几种case:
     * a. 如果timeout为null，则采用tryGet方式，即时获取
     * b. 如果timeout不为null
     *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
     *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
     * 
     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
     * </pre>
     */
    @Override
    public Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
                                                                                                 throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        checkSubscribe(clientIdentity);
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        synchronized (canalInstance) {
            // 获取到流式数据中的最后一批获取的位置
            PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().getLastestBatch(clientIdentity);

            if (positionRanges != null) {
                throw new CanalServerException(String.format("clientId:%s has last batch:[%s] isn't ack , maybe loss data",
                    clientIdentity.getClientId(),
                    positionRanges));
            }

            Events<Event> events = null;
            Position start = canalInstance.getMetaManager().getCursor(clientIdentity);
            events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);

            if (CollectionUtils.isEmpty(events.getEvents())) {
                logger.debug("get successfully, clientId:{} batchSize:{} but result is null",
                    clientIdentity.getClientId(),
                    batchSize);
                return new Message(-1, true, new ArrayList()); // 返回空包，避免生成batchId，浪费性能
            } else {
                // 记录到流式信息
                Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange());
                boolean raw = isRaw(canalInstance.getEventStore());
                List entrys = null;
                if (raw) {
                    // new list
                    entrys = events.getEvents().stream().map(Event::getRawEntry).collect(Collectors.toList());
                } else {
                    entrys = events.getEvents().stream().map(Event::getEntry).collect(Collectors.toList());
                }
                if (logger.isInfoEnabled()) {
                    logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
                        clientIdentity.getClientId(),
                        batchSize,
                        entrys.size(),
                        batchId,
                        events.getPositionRange());
                }
                // 直接提交ack
                ack(clientIdentity, batchId);
                return new Message(batchId, raw, entrys);
            }
        }
    }

    /**
     * 不指定 position 获取事件。canal 会记住此 client 最新的 position。 <br/>
     * 如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。
     *
     * <pre>
     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
     * </pre>
     */
    @Override
    public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize) throws CanalServerException {
        return getWithoutAck(clientIdentity, batchSize, null, null);
    }

    /**
     * 不指定 position 获取事件。canal 会记住此 client 最新的 position。 <br/>
     * 如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。
     *
     * <pre>
     * 几种case:
     * a. 如果timeout为null，则采用tryGet方式，即时获取
     * b. 如果timeout不为null
     *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
     *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
     * 
     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
     * </pre>
     */
    @Override
    public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
                                                                                                           throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        checkSubscribe(clientIdentity);

        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        synchronized (canalInstance) {
            // 获取到流式数据中的最后一批获取的位置
            PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().getLastestBatch(clientIdentity);

            Events<Event> events = null;
            if (positionRanges != null) { // 存在流数据
                events = getEvents(canalInstance.getEventStore(), positionRanges.getStart(), batchSize, timeout, unit);
            } else {// ack后第一次获取
                Position start = canalInstance.getMetaManager().getCursor(clientIdentity);
                if (start == null) { // 第一次，还没有过ack记录，则获取当前store中的第一条
                    start = canalInstance.getEventStore().getFirstPosition();
                }

                events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);
            }

            if (CollectionUtils.isEmpty(events.getEvents())) {
                // logger.debug("getWithoutAck successfully, clientId:{}
                // batchSize:{} but result
                // is null",
                // clientIdentity.getClientId(),
                // batchSize);
                return new Message(-1, true, new ArrayList()); // 返回空包，避免生成batchId，浪费性能
            } else {
                // 记录到流式信息
                Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange());
                boolean raw = isRaw(canalInstance.getEventStore());
                List entrys = null;
                if (raw) {
                    // new list
                    entrys = events.getEvents().stream().map(Event::getRawEntry).collect(Collectors.toList());
                } else {
                    entrys = events.getEvents().stream().map(Event::getEntry).collect(Collectors.toList());
                }
                if (logger.isInfoEnabled()) {
                    logger.info("getWithoutAck successfully, clientId:{} batchSize:{}  real size is {} and result is [batchId:{} , position:{}]",
                        clientIdentity.getClientId(),
                        batchSize,
                        entrys.size(),
                        batchId,
                        events.getPositionRange());
                }
                return new Message(batchId, raw, entrys);
            }

        }
    }

    /**
     * 查询当前未被ack的batch列表，batchId会按照从小到大进行返回
     */
    public List<Long> listBatchIds(ClientIdentity clientIdentity) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        checkSubscribe(clientIdentity);

        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        Map<Long, PositionRange> batchs = canalInstance.getMetaManager().listAllBatchs(clientIdentity);
        List<Long> result = new ArrayList<>(batchs.keySet());
        Collections.sort(result);
        return result;
    }

    /**
     * 进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
     *
     * <pre>
     * 注意：进行反馈时必须按照batchId的顺序进行ack(需有客户端保证)
     * </pre>
     */
    @Override
    public void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        checkSubscribe(clientIdentity);

        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        PositionRange<LogPosition> positionRanges = null;
        positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity, batchId); // 更新位置
        if (positionRanges == null) { // 说明是重复的ack/rollback
            throw new CanalServerException(String.format("ack error , clientId:%s batchId:%d is not exist , please check",
                clientIdentity.getClientId(),
                batchId));
        }

        // 更新cursor最好严格判断下位置是否有跳跃更新
        // Position position = lastRollbackPostions.get(clientIdentity);
        // if (position != null) {
        // // Position position =
        // canalInstance.getMetaManager().getCursor(clientIdentity);
        // LogPosition minPosition =
        // CanalEventUtils.min(positionRanges.getStart(), (LogPosition)
        // position);
        // if (minPosition == position) {// ack的position要晚于该最后ack的位置，可能有丢数据
        // throw new CanalServerException(
        // String.format(
        // "ack error , clientId:%s batchId:%d %s is jump ack , last ack:%s",
        // clientIdentity.getClientId(), batchId, positionRanges,
        // position));
        // }
        // }

        // 更新cursor
        if (positionRanges.getAck() != null) {
            canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck());
            if (logger.isInfoEnabled()) {
                logger.info("ack successfully, clientId:{} batchId:{} position:{}",
                    clientIdentity.getClientId(),
                    batchId,
                    positionRanges);
            }
        }

        // 可定时清理数据
        canalInstance.getEventStore().ack(positionRanges.getEnd(), positionRanges.getEndSeq());
    }

    /**
     * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
     */
    @Override
    public void rollback(ClientIdentity clientIdentity) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
        if (!hasSubscribe) {
            return;
        }

        synchronized (canalInstance) {
            // 清除batch信息
            canalInstance.getMetaManager().clearAllBatchs(clientIdentity);
            // rollback eventStore中的状态信息
            canalInstance.getEventStore().rollback();
            logger.info("rollback successfully, clientId:{}", new Object[] { clientIdentity.getClientId() });
        }
    }

    /**
     * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
     */
    @Override
    public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException {
        checkStart(clientIdentity.getDestination());
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());

        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
        if (!hasSubscribe) {
            return;
        }
        synchronized (canalInstance) {
            // 清除batch信息
            PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity,
                batchId);
            if (positionRanges == null) { // 说明是重复的ack/rollback
                throw new CanalServerException(String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
                    clientIdentity.getClientId(),
                    batchId));
            }

            // lastRollbackPostions.put(clientIdentity,
            // positionRanges.getEnd());// 记录一下最后rollback的位置
            // TODO 后续rollback到指定的batchId位置
            canalInstance.getEventStore().rollback();// rollback
                                                     // eventStore中的状态信息
            logger.info("rollback successfully, clientId:{} batchId:{} position:{}",
                clientIdentity.getClientId(),
                batchId,
                positionRanges);
        }
    }

    public Map<String, CanalInstance> getCanalInstances() {
        return Maps.newHashMap(canalInstances);
    }

    // ======================== helper method =======================

    /**
     * 根据不同的参数，选择不同的方式获取数据
     */
    private Events<Event> getEvents(CanalEventStore eventStore, Position start, int batchSize, Long timeout,
                                    TimeUnit unit) {
        if (timeout == null) {
            return eventStore.tryGet(start, batchSize);
        } else {
            try {
                if (timeout <= 0) {
                    return eventStore.get(start, batchSize);
                } else {
                    return eventStore.get(start, batchSize, timeout, unit);
                }
            } catch (Exception e) {
                throw new CanalServerException(e);
            }
        }
    }

    private void checkSubscribe(ClientIdentity clientIdentity) {
        CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
        boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
        if (!hasSubscribe) {
            throw new CanalServerException(String.format("ClientIdentity:%s should subscribe first",
                clientIdentity.toString()));
        }
    }

    private void checkStart(String destination) {
        if (!isStart(destination)) {
            throw new CanalServerException(String.format("destination:%s should start first", destination));
        }
    }

    private void loadCanalMetrics() {
        ServiceLoader<CanalMetricsProvider> providers = ServiceLoader.load(CanalMetricsProvider.class);
        List<CanalMetricsProvider> list = new ArrayList<>();
        for (CanalMetricsProvider provider : providers) {
            list.add(provider);
        }

        if (list.isEmpty()) {
            return;
        }

        // only allow ONE provider
        if (list.size() > 1) {
            logger.warn("Found more than one CanalMetricsProvider, use the first one.");
            // 报告冲突
            for (CanalMetricsProvider p : list) {
                logger.warn("Found CanalMetricsProvider: {}.", p.getClass().getName());
            }
        }

        CanalMetricsProvider provider = list.get(0);
        this.metrics = provider.getService();
    }

    private boolean isRaw(CanalEventStore eventStore) {
        if (eventStore instanceof MemoryEventStoreWithBuffer) {
            return ((MemoryEventStoreWithBuffer) eventStore).isRaw();
        }

        return true;
    }

    // ========= setter ==========

    public void setCanalInstanceGenerator(CanalInstanceGenerator canalInstanceGenerator) {
        this.canalInstanceGenerator = canalInstanceGenerator;
    }

    public void setMetricsPort(int metricsPort) {
        this.metricsPort = metricsPort;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public void setPasswd(String passwd) {
        this.passwd = passwd;
    }

}

@Ignore
public abstract class BaseCanalServerWithEmbededTest {

    protected static final String   cluster1       = "127.0.0.1:2188";
    protected static final String   DESTINATION    = "example";
    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       = "canal";
    protected static final String   PASSWORD       = "canal";
    protected static final String   FILTER         = ".*\\\\..*";

    private CanalServerWithEmbedded server;
    private ClientIdentity          clientIdentity = new ClientIdentity(DESTINATION, (short) 1);                               ;

    @Before
    public void setUp() {
        server = CanalServerWithEmbedded.instance();
        server.setCanalInstanceGenerator(destination -> {
            Canal canal = buildCanal();
            return new CanalInstanceWithManager(canal, FILTER);
        });
        server.start();
        server.start(DESTINATION);
    }

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

    @Test
    public void testGetWithoutAck() {
        int maxEmptyCount = 10;
        int emptyCount = 0;
        int totalCount = 0;
        server.subscribe(clientIdentity);
        while (emptyCount < maxEmptyCount) {
            Message message = server.getWithoutAck(clientIdentity, 11);
            if (CollectionUtils.isEmpty(message.getEntries())) {
                emptyCount++;
                try {
                    Thread.sleep(emptyCount * 300L);
                } catch (InterruptedException e) {
                    Assert.fail();
                }

                System.out.println("empty count : " + emptyCount);
            } else {
                emptyCount = 0;
                totalCount += message.getEntries().size();
                server.ack(clientIdentity, message.getId());
            }
        }

        System.out.println("!!!!!! testGetWithoutAck totalCount : " + totalCount);
        server.unsubscribe(clientIdentity);
    }

    @Test
    public void testGet() {
        int maxEmptyCount = 10;
        int emptyCount = 0;
        int totalCount = 0;
        server.subscribe(clientIdentity);
        while (emptyCount < maxEmptyCount) {
            Message message = server.get(clientIdentity, 11);
            if (CollectionUtils.isEmpty(message.getEntries())) {
                emptyCount++;
                try {
                    Thread.sleep(emptyCount * 300L);
                } catch (InterruptedException e) {
                    Assert.fail();
                }

                System.out.println("empty count : " + emptyCount);
            } else {
                emptyCount = 0;
                totalCount += message.getEntries().size();
            }
        }

        System.out.println("!!!!!! testGet totalCount : " + totalCount);
        server.unsubscribe(clientIdentity);
    }
}

public interface CanalMetricsService {

    /**
     * Initialization on canal server startup.
     */
    void initialize();

    /**
     * Clean-up at canal server stop phase.
     */
    void terminate();

    /**
     * @return {@code true} if the metrics service is running, otherwise {@code false}.
     */
    boolean isRunning();

    /**
     * Register instance level metrics for specified instance.
     * @param instance {@link CanalInstance}
     */
    void register(CanalInstance instance);

    /**
     * Unregister instance level metrics for specified instance.
     * @param instance {@link CanalInstance}
     */
    void unregister(CanalInstance instance);

    /**
     * @param port server port for pull
     */
    void setServerPort(int port);

}

public class NopCanalMetricsService implements CanalMetricsService {

    public static final NopCanalMetricsService NOP = new NopCanalMetricsService();

    private NopCanalMetricsService() {}

    @Override
    public void initialize() {

    }

    @Override
    public void terminate() {

    }

    @Override
    public boolean isRunning() {
        return false;
    }

    @Override
    public void register(CanalInstance instance) {

    }

    @Override
    public void unregister(CanalInstance instance) {

    }

    @Override
    public void setServerPort(int port) {

    }
}

public interface CanalMetricsProvider {

    /**
     * @return Impl of {@link CanalMetricsService}
     */
    CanalMetricsService getService();

}