ShardingSphere~6

Scaling

这一部分的实现跟之前有显著的不同，先从入口shardingsphere-scaling-bootstrap ScalingServerBootstrap 看下，就是先初始化配置，再启动一个netty服务

@NoArgsConstructor(access = AccessLevel.PRIVATE)
@Slf4j
public final class ScalingServerBootstrap {
    public static void main(final String[] args) {
        // CHECKSTYLE:ON
        log.info("ShardingSphere-Scaling Server Startup");
        ServerConfigurationInitializer.init();
        startScalingServer();
    }
    
    @SneakyThrows(InterruptedException.class)
    private static void startScalingServer() {
        log.info("Start scaling server");
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new HttpServerInitializer());
            int port = ScalingContext.getInstance().getServerConfig().getPort();
            Channel channel = bootstrap.bind(port).sync().channel();
            log.info("ShardingSphere-Scaling is server on http://127.0.0.1:{}/", port);
            channel.closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

继续跟进，看起来似乎就是解码http请求，再处理下

public final class HttpServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(final SocketChannel socketChannel) {
        ChannelPipeline channelPipeline = socketChannel.pipeline();
        channelPipeline.addLast(new HttpServerCodec());
        channelPipeline.addLast(new HttpObjectAggregator(65536));
        channelPipeline.addLast(new HttpServerHandler());
    }
}

进到最后一个handler，就十分清晰了，得到请求后，按照路径对应执行不同的/scaling/job/***

public final class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    
    private static final Gson GSON = new GsonBuilder().disableHtmlEscaping().serializeNulls().setLongSerializationPolicy(LongSerializationPolicy.STRING).create();
    
    private final ScalingAPI scalingAPI = ScalingAPIFactory.getScalingAPI();
    
    @Override
    protected void channelRead0(final ChannelHandlerContext context, final FullHttpRequest request) {
        String requestPath = request.uri().toLowerCase();
        String requestBody = request.content().toString(CharsetUtil.UTF_8);
        log.info("Http request path: {}", requestPath);
        log.info("Http request body: {}", requestBody);
        if ("/scaling/job/start".equalsIgnoreCase(requestPath) && request.method().equals(HttpMethod.POST)) {
            startJob(context, requestBody);
            return;
        }
        if ("/scaling/job/list".equalsIgnoreCase(requestPath)) {
            listJobs(context);
            return;
        }
        if (requestPath.startsWith("/scaling/job/progress/")) {
            getJobProgress(context, requestPath);
            return;
        }
        if (requestPath.startsWith("/scaling/job/stop/")) {
            stopJob(context, requestPath);
            return;
        }
        if (requestPath.contains("/scaling/job/check/")) {
            checkJob(context, requestPath);
            return;
        }
        if (requestPath.contains("/scaling/job/reset/")) {
            resetJob(context, requestPath);
            return;
        }
        response(ResponseContentUtil.handleBadRequest("Not support request!"), context, HttpResponseStatus.BAD_REQUEST);
    }
    
    private void startJob(final ChannelHandlerContext context, final String requestBody) {
        Optional<Long> jobId = scalingAPI.start(GSON.fromJson(requestBody, JobConfiguration.class));
        if (jobId.isPresent()) {
            response(ResponseContentUtil.build(jobId.get()), context, HttpResponseStatus.OK);
            return;
        }
        response(ResponseContentUtil.handleBadRequest("Invalid scaling job config!"), context, HttpResponseStatus.BAD_REQUEST);
    }
    
    private void listJobs(final ChannelHandlerContext context) {
        List<JobInfo> jobContexts = scalingAPI.list();
        response(ResponseContentUtil.build(jobContexts), context, HttpResponseStatus.OK);
    }
    
    private void getJobProgress(final ChannelHandlerContext context, final String requestPath) {
        try {
            response(ResponseContentUtil.build(scalingAPI.getProgress(getJobId(requestPath))), context, HttpResponseStatus.OK);
        } catch (final ScalingJobNotFoundException ex) {
            response(ResponseContentUtil.handleBadRequest(ex.getMessage()), context, HttpResponseStatus.BAD_REQUEST);
        }
    }
    
    private void stopJob(final ChannelHandlerContext context, final String requestPath) {
        scalingAPI.stop(getJobId(requestPath));
        response(ResponseContentUtil.success(), context, HttpResponseStatus.OK);
    }
    
    private void checkJob(final ChannelHandlerContext context, final String requestPath) {
        try {
            response(ResponseContentUtil.build(scalingAPI.dataConsistencyCheck(getJobId(requestPath))), context, HttpResponseStatus.OK);
        } catch (final ScalingJobNotFoundException ex) {
            response(ResponseContentUtil.handleBadRequest(ex.getMessage()), context, HttpResponseStatus.BAD_REQUEST);
        }
    }
    
    private void resetJob(final ChannelHandlerContext context, final String requestPath) {
        try {
            scalingAPI.reset(getJobId(requestPath));
            response(ResponseContentUtil.success(), context, HttpResponseStatus.OK);
        } catch (final ScalingJobNotFoundException | SQLException ex) {
            response(ResponseContentUtil.handleBadRequest(ex.getMessage()), context, HttpResponseStatus.BAD_REQUEST);
        }
    }
    
    private long getJobId(final String requestPath) {
        return Long.parseLong(requestPath.split("/")[4]);
    }
    
    private void response(final Object content, final ChannelHandlerContext context, final HttpResponseStatus status) {
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status, Unpooled.copiedBuffer(GSON.toJson(content), CharsetUtil.UTF_8));
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain;charset=UTF-8");
        HttpUtil.setContentLength(response, response.content().readableBytes());
        response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        context.writeAndFlush(response);
    }
    
    @Override
    public void exceptionCaught(final ChannelHandlerContext context, final Throwable cause) {
        log.error("Http request handle occur error:", cause);
        response(ResponseContentUtil.handleException(cause.toString()), context, HttpResponseStatus.INTERNAL_SERVER_ERROR);
        context.close();
    }
}

具体执行已经脱离了netty应用本身，这里shardingsphere框架封装了一个ScalingAPIFactory

具体看这个类的内容，发现实际执行是依赖了ElasticJob，所有命令都相当于是注册中心协调下的job运行

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class ScalingAPIFactory {
    
    /**
     * Get scaling API.
     *
     * @return scaling API
     */
    public static ScalingAPI getScalingAPI() {
        return ScalingAPIHolder.getInstance();
    }
    
    /**
     * Get governance repository API.
     *
     * @return governance repository API
     */
    public static GovernanceRepositoryAPI getGovernanceRepositoryAPI() {
        return GovernanceRepositoryAPIHolder.getInstance();
    }
    
    /**
     * Get job statistics API.
     *
     * @return job statistics API
     */
    public static JobStatisticsAPI getJobStatisticsAPI() {
        return ElasticJobAPIHolder.getInstance().getJobStatisticsAPI();
    }
    
    /**
     * Get job configuration API.
     *
     * @return job configuration API
     */
    public static JobConfigurationAPI getJobConfigurationAPI() {
        return ElasticJobAPIHolder.getInstance().getJobConfigurationAPI();
    }
    
    /**
     * Get job operate API.
     *
     * @return job operate API
     */
    public static JobOperateAPI getJobOperateAPI() {
        return ElasticJobAPIHolder.getInstance().getJobOperateAPI();
    }
    
    /**
     * Get registry center.
     *
     * @return Coordinator registry center
     */
    public static CoordinatorRegistryCenter getRegistryCenter() {
        return RegistryCenterHolder.getInstance();
    }
    
    private static void checkServerConfig() {
        ServerConfiguration serverConfig = ScalingContext.getInstance().getServerConfig();
        Preconditions.checkNotNull(serverConfig, "Scaling server configuration is required.");
        Preconditions.checkNotNull(serverConfig.getGovernanceConfig(), "Governance configuration is required.");
    }
    
    private static final class ScalingAPIHolder {
        
        private static volatile ScalingAPI instance;
        
        public static ScalingAPI getInstance() {
            if (null == instance) {
                synchronized (ScalingAPIFactory.class) {
                    if (null == instance) {
                        checkServerConfig();
                        instance = new ScalingAPIImpl();
                    }
                }
            }
            return instance;
        }
    }
    
    private static final class GovernanceRepositoryAPIHolder {
        
        private static volatile GovernanceRepositoryAPI instance;
        
        static {
            ShardingSphereServiceLoader.register(RegistryCenterRepository.class);
        }
        
        public static GovernanceRepositoryAPI getInstance() {
            if (null == instance) {
                synchronized (ScalingAPIFactory.class) {
                    if (null == instance) {
                        instance = createGovernanceRepositoryAPI();
                    }
                }
            }
            return instance;
        }
        
        private static GovernanceRepositoryAPI createGovernanceRepositoryAPI() {
            checkServerConfig();
            GovernanceConfiguration governanceConfig = ScalingContext.getInstance().getServerConfig().getGovernanceConfig();
            RegistryCenterConfiguration registryCenterConfig = governanceConfig.getRegistryCenterConfiguration();
            RegistryCenterRepository registryCenterRepository = TypedSPIRegistry.getRegisteredService(RegistryCenterRepository.class, registryCenterConfig.getType(), registryCenterConfig.getProps());
            registryCenterRepository.init(governanceConfig.getName(), registryCenterConfig);
            return new GovernanceRepositoryAPIImpl(registryCenterRepository);
        }
    }
    
    @Getter
    private static final class ElasticJobAPIHolder {
        
        private static volatile ElasticJobAPIHolder instance;
        
        private final JobStatisticsAPI jobStatisticsAPI;
        
        private final JobConfigurationAPI jobConfigurationAPI;
        
        private final JobOperateAPI jobOperateAPI;
        
        private ElasticJobAPIHolder() {
            checkServerConfig();
            GovernanceConfiguration governanceConfig = ScalingContext.getInstance().getServerConfig().getGovernanceConfig();
            String namespace = governanceConfig.getName() + ScalingConstant.SCALING_ROOT;
            jobStatisticsAPI = JobAPIFactory.createJobStatisticsAPI(governanceConfig.getRegistryCenterConfiguration().getServerLists(), namespace, null);
            jobConfigurationAPI = JobAPIFactory.createJobConfigurationAPI(governanceConfig.getRegistryCenterConfiguration().getServerLists(), namespace, null);
            jobOperateAPI = JobAPIFactory.createJobOperateAPI(governanceConfig.getRegistryCenterConfiguration().getServerLists(), namespace, null);
        }
        
        public static ElasticJobAPIHolder getInstance() {
            if (null == instance) {
                synchronized (ScalingAPIFactory.class) {
                    if (null == instance) {
                        instance = new ElasticJobAPIHolder();
                    }
                }
            }
            return instance;
        }
    }
    
    private static final class RegistryCenterHolder {
        
        private static volatile CoordinatorRegistryCenter instance;
        
        public static CoordinatorRegistryCenter getInstance() {
            if (null == instance) {
                synchronized (ScalingAPIFactory.class) {
                    if (null == instance) {
                        instance = createRegistryCenter();
                    }
                }
            }
            return instance;
        }
        
        private static CoordinatorRegistryCenter createRegistryCenter() {
            CoordinatorRegistryCenter result = new ZookeeperRegistryCenter(getZookeeperConfig());
            result.init();
            return result;
        }
        
        private static ZookeeperConfiguration getZookeeperConfig() {
            checkServerConfig();
            GovernanceConfiguration governanceConfig = ScalingContext.getInstance().getServerConfig().getGovernanceConfig();
            ZookeeperConfiguration result = new ZookeeperConfiguration(governanceConfig.getRegistryCenterConfiguration().getServerLists(),
                    governanceConfig.getName() + ScalingConstant.SCALING_ROOT);
            Properties props = governanceConfig.getRegistryCenterConfiguration().getProps();
            result.setMaxSleepTimeMilliseconds(getProperty(props, "max.sleep.time.milliseconds", result.getMaxSleepTimeMilliseconds()));
            result.setBaseSleepTimeMilliseconds(getProperty(props, "base.sleep.time.milliseconds", result.getBaseSleepTimeMilliseconds()));
            result.setConnectionTimeoutMilliseconds(getProperty(props, "connection.timeout.milliseconds", result.getConnectionTimeoutMilliseconds()));
            result.setSessionTimeoutMilliseconds(getProperty(props, "session.timeout.milliseconds", result.getSessionTimeoutMilliseconds()));
            return result;
        }
        
        private static int getProperty(final Properties props, final String key, final int defaultValue) {
            return Strings.isNullOrEmpty(props.getProperty(key)) ? defaultValue : Integer.parseInt(props.getProperty(key));
        }
    }
}

再看下包里的另一个文件 ScalingWorkerBootstrap，也是个有main方法的类，看名字先初始化配置再初始化工作者

@NoArgsConstructor(access = AccessLevel.PRIVATE)
@Slf4j
public final class ScalingWorkerBootstrap {
    public static void main(final String[] args) {
        // CHECKSTYLE:ON
        log.info("ShardingSphere-Scaling Worker Startup");
        ServerConfigurationInitializer.init();
        ScalingWorker.init();
        wait0();
    }
    
    private static synchronized void wait0() {
        try {
            ScalingWorkerBootstrap.class.wait();
        } catch (final InterruptedException ignored) {
        }
    }
}

工作者是做什么工作的？通过事件、定时任务、注册中心驱动整个job的执行周期

@Slf4j
public final class ScalingWorker {
    
    private static final ScalingWorker INSTANCE = new ScalingWorker();
    
    private final ScalingAPI scalingAPI = ScalingAPIFactory.getScalingAPI();
    
    /**
     * Init scaling worker.
     */
    public static void init() {
        ShardingSphereEventBus.getInstance().register(INSTANCE);
        new FinishedCheckJobExecutor().start();
        new ScalingJobExecutor().start();
    }
    
    /**
     * Start scaling job.
     *
     * @param event start scaling event.
     */
    @Subscribe
    public void start(final StartScalingEvent event) {
        log.info("Start scaling job by {}", event);
        Optional<Long> jobId = scalingAPI.start(createJobConfig(event));
        if (!jobId.isPresent()) {
            log.info("Switch rule configuration ruleCacheId = {} immediately.", event.getRuleCacheId());
            ShardingSphereEventBus.getInstance().post(new SwitchRuleConfigurationEvent(event.getSchemaName(), event.getRuleCacheId()));
        }
    }
    
    private JobConfiguration createJobConfig(final StartScalingEvent event) {
        JobConfiguration result = new JobConfiguration();
        result.setRuleConfig(getRuleConfiguration(event));
        result.setHandleConfig(new HandleConfiguration(new WorkflowConfiguration(event.getSchemaName(), event.getRuleCacheId())));
        return result;
    }
    
    private RuleConfiguration getRuleConfiguration(final StartScalingEvent event) {
        RuleConfiguration result = new RuleConfiguration();
        result.setSource(new ShardingSphereJDBCDataSourceConfiguration(event.getSourceDataSource(), event.getSourceRule()).wrap());
        result.setTarget(new ShardingSphereJDBCDataSourceConfiguration(event.getTargetDataSource(), event.getTargetRule()).wrap());
        return result;
    }
}

ScalingJobExecutor 监听注册中心任务，收到通知后做对应的执行动作，相当于通过注册中心解耦了http请求接收和真正的执行处理

@Slf4j
public final class ScalingJobExecutor extends AbstractScalingExecutor {
    
    private static final Pattern CONFIG_PATTERN = Pattern.compile(ScalingConstant.SCALING_ROOT + "/(\d+)/config");
    
    private static final Set<String> EXECUTING_JOBS = Sets.newConcurrentHashSet();
    
    @Override
    public void start() {
        super.start();
        log.info("Start scaling job executor.");
        watchGovernanceRepositoryConfiguration();
    }
    
    private void watchGovernanceRepositoryConfiguration() {
        ScalingAPIFactory.getGovernanceRepositoryAPI().watch(ScalingConstant.SCALING_ROOT, event -> {
            Optional<JobConfigurationPOJO> jobConfigPOJOOptional = getJobConfigPOJO(event);
            if (!jobConfigPOJOOptional.isPresent()) {
                return;
            }
            JobConfigurationPOJO jobConfigPOJO = jobConfigPOJOOptional.get();
            if (DataChangedEvent.Type.DELETED == event.getType() || jobConfigPOJO.isDisabled()) {
                EXECUTING_JOBS.remove(jobConfigPOJO.getJobName());
                JobSchedulerCenter.stop(Long.parseLong(jobConfigPOJO.getJobName()));
                return;
            }
            switch (event.getType()) {
                case ADDED:
                case UPDATED:
                    execute(jobConfigPOJO);
                    break;
                default:
                    break;
            }
        });
    }
    
    private Optional<JobConfigurationPOJO> getJobConfigPOJO(final DataChangedEvent event) {
        try {
            if (CONFIG_PATTERN.matcher(event.getKey()).matches()) {
                log.info("{} job config: {} = {}", event.getType(), event.getKey(), event.getValue());
                return Optional.of(YamlEngine.unmarshal(event.getValue(), JobConfigurationPOJO.class));
            }
            // CHECKSTYLE:OFF
        } catch (final Exception ex) {
            // CHECKSTYLE:ON
            log.error("analyze job config pojo failed.", ex);
        }
        return Optional.empty();
    }
    
    private void execute(final JobConfigurationPOJO jobConfigPOJO) {
        if (EXECUTING_JOBS.add(jobConfigPOJO.getJobName())) {
            new OneOffJobBootstrap(ScalingAPIFactory.getRegistryCenter(), new ScalingJob(), jobConfigPOJO.toJobConfiguration()).execute();
        }
    }
}

具体的扩展操作是分为几个步骤的，官网写的不错，转

准备阶段：在准备阶段，弹性伸缩模块会进行数据源连通性及权限的校验，同时进行存量数据的统计、日志位点的记录，最后根据数据量和用户设置的并行度，对任务进行分片。

存量数据迁移阶段：执行在准备阶段拆分好的存量数据迁移作业，存量迁移阶段采用 JDBC 查询的方式，直接从数据节点中读取数据，并使用新规则写入到新集群中。

增量数据同步阶段：由于存量数据迁移耗费的时间受到数据量和并行度等因素影响，此时需要对这段时间内业务新增的数据进行同步。 不同的数据库使用的技术细节不同，但总体上均为基于复制协议或 WAL 日志实现的变更数据捕获功能。

• MySQL：订阅并解析 binlog

• PostgreSQL：采用官方逻辑复制

这些捕获的增量数据，同样会由弹性伸缩模块根据新规则写入到新数据节点中。当增量数据基本同步完成时（由于业务系统未停止，增量数据是不断的），则进入规则切换阶段。

规则切换阶段：在此阶段，可能存在一定时间的业务只读窗口期，通过设置数据库只读或ShardingSphere的熔断机制，让旧数据节点中的数据短暂静态，确保增量同步已完全完成。这个窗口期时间短则数秒，长则数分钟，取决于数据量和用户是否需要对数据进行强校验。 确认完成后，Apache ShardingSphere 可通过配置中心修改配置，将业务导向新规则的集群，弹性伸缩完成。

先不论其他，先来看下MySql Binlog同步相关代码

MySQLClient 这个类整体就是启动一个Netty服务，支持与MYSQL做协议交互，伪装从节点做binlog同步，dump等。。。

@RequiredArgsConstructor
@Slf4j
public final class MySQLClient {
    
    private final ConnectInfo connectInfo;
    
    private EventLoopGroup eventLoopGroup;
    
    private Channel channel;
    
    private Promise<Object> responseCallback;
    
    private final ArrayBlockingQueue<AbstractBinlogEvent> blockingEventQueue = new ArrayBlockingQueue<>(10000);
    
    private ServerInfo serverInfo;
    
    /**
     * Connect to MySQL.
     */
    public synchronized void connect() {
        eventLoopGroup = new NioEventLoopGroup(1);
        responseCallback = new DefaultPromise<>(eventLoopGroup.next());
        channel = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.AUTO_READ, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(final SocketChannel socketChannel) {
                        socketChannel.pipeline().addLast(new PacketCodec(new MySQLPacketCodecEngine()));
                        socketChannel.pipeline().addLast(new MySQLNegotiatePackageDecoder());
                        socketChannel.pipeline().addLast(new MySQLCommandPacketDecoder());
                        socketChannel.pipeline().addLast(new MySQLNegotiateHandler(connectInfo.getUsername(), connectInfo.getPassword(), responseCallback));
                        socketChannel.pipeline().addLast(new MySQLCommandResponseHandler());
                    }
                }).connect(connectInfo.getHost(), connectInfo.getPort()).channel();
        serverInfo = waitExpectedResponse(ServerInfo.class);
    }
    
    /**
     * Execute command.
     *
     * @param queryString query string
     * @return true if execute successfully, otherwise false
     */
    public synchronized boolean execute(final String queryString) {
        responseCallback = new DefaultPromise<>(eventLoopGroup.next());
        MySQLComQueryPacket comQueryPacket = new MySQLComQueryPacket(queryString);
        channel.writeAndFlush(comQueryPacket);
        return null != waitExpectedResponse(MySQLOKPacket.class);
    }
    
    /**
     * Execute update.
     *
     * @param queryString query string
     * @return affected rows
     */
    public synchronized int executeUpdate(final String queryString) {
        responseCallback = new DefaultPromise<>(eventLoopGroup.next());
        MySQLComQueryPacket comQueryPacket = new MySQLComQueryPacket(queryString);
        channel.writeAndFlush(comQueryPacket);
        return (int) Objects.requireNonNull(waitExpectedResponse(MySQLOKPacket.class)).getAffectedRows();
    }
    
    /**
     * Execute query.
     *
     * @param queryString query string
     * @return result set
     */
    public synchronized InternalResultSet executeQuery(final String queryString) {
        responseCallback = new DefaultPromise<>(eventLoopGroup.next());
        MySQLComQueryPacket comQueryPacket = new MySQLComQueryPacket(queryString);
        channel.writeAndFlush(comQueryPacket);
        return waitExpectedResponse(InternalResultSet.class);
    }
    
    /**
     * Start dump binlog.
     *
     * @param binlogFileName binlog file name
     * @param binlogPosition binlog position
     */
    public synchronized void subscribe(final String binlogFileName, final long binlogPosition) {
        initDumpConnectSession();
        registerSlave();
        dumpBinlog(binlogFileName, binlogPosition, queryChecksumLength());
    }
    
    private void initDumpConnectSession() {
        if (serverInfo.getServerVersion().greaterThanOrEqualTo(5, 6, 0)) {
            execute("SET @MASTER_BINLOG_CHECKSUM= @@GLOBAL.BINLOG_CHECKSUM");
        }
    }
    
    private void registerSlave() {
        responseCallback = new DefaultPromise<>(eventLoopGroup.next());
        InetSocketAddress localAddress = (InetSocketAddress) channel.localAddress();
        MySQLComRegisterSlaveCommandPacket packet = new MySQLComRegisterSlaveCommandPacket(
                connectInfo.getServerId(), localAddress.getHostName(), connectInfo.getUsername(), connectInfo.getPassword(), localAddress.getPort());
        channel.writeAndFlush(packet);
        waitExpectedResponse(MySQLOKPacket.class);
    }
    
    private int queryChecksumLength() {
        if (!serverInfo.getServerVersion().greaterThanOrEqualTo(5, 6, 0)) {
            return 0;
        }
        InternalResultSet resultSet = executeQuery("SELECT @@GLOBAL.BINLOG_CHECKSUM");
        String checksumType = resultSet.getFieldValues().get(0).getData().iterator().next().toString();
        switch (checksumType) {
            case "None":
                return 0;
            case "CRC32":
                return 4;
            default:
                throw new UnsupportedOperationException(checksumType);
        }
    }
    
    private void dumpBinlog(final String binlogFileName, final long binlogPosition, final int checksumLength) {
        responseCallback = null;
        channel.pipeline().remove(MySQLCommandPacketDecoder.class);
        channel.pipeline().remove(MySQLCommandResponseHandler.class);
        channel.pipeline().addLast(new MySQLBinlogEventPacketDecoder(checksumLength));
        channel.pipeline().addLast(new MySQLBinlogEventHandler());
        channel.writeAndFlush(new MySQLComBinlogDumpCommandPacket((int) binlogPosition, connectInfo.getServerId(), binlogFileName));
    }
    
    /**
     * Poll binlog event.
     *
     * @return binlog event
     */
    public synchronized AbstractBinlogEvent poll() {
        try {
            return blockingEventQueue.poll(100, TimeUnit.MILLISECONDS);
        } catch (final InterruptedException ignored) {
            return null;
        }
    }
    
    @SuppressWarnings("unchecked")
    private <T> T waitExpectedResponse(final Class<T> type) {
        try {
            Object response = responseCallback.get();
            if (null == response) {
                return null;
            }
            if (type.equals(response.getClass())) {
                return (T) response;
            }
            if (response instanceof MySQLErrPacket) {
                throw new RuntimeException(((MySQLErrPacket) response).getErrorMessage());
            }
            throw new RuntimeException("unexpected response type");
        } catch (final InterruptedException | ExecutionException ex) {
            throw new RuntimeException(ex);
        }
    }
    
    private final class MySQLCommandResponseHandler extends ChannelInboundHandlerAdapter {
        
        @Override
        public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
            if (null != responseCallback) {
                responseCallback.setSuccess(msg);
            }
        }
        
        @Override
        public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
            if (null != responseCallback) {
                responseCallback.setFailure(cause);
                log.error("protocol resolution error", cause);
            }
        }
    }
    
    private final class MySQLBinlogEventHandler extends ChannelInboundHandlerAdapter {
        
        private AbstractBinlogEvent lastBinlogEvent;
        
        @Override
        public void channelRead(final ChannelHandlerContext ctx, final Object msg) throws Exception {
            if (msg instanceof AbstractBinlogEvent) {
                lastBinlogEvent = (AbstractBinlogEvent) msg;
                blockingEventQueue.put(lastBinlogEvent);
            }
        }
        
        @Override
        public void channelInactive(final ChannelHandlerContext ctx) {
            log.warn("channel inactive");
            reconnect();
        }
        
        @Override
        public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
            log.error("protocol resolution error", cause);
            reconnect();
        }
        
        private void reconnect() {
            log.info("reconnect mysql client.");
            closeOldChannel();
            connect();
            subscribe(lastBinlogEvent.getFileName(), lastBinlogEvent.getPosition());
        }
        
        private void closeOldChannel() {
            try {
                channel.closeFuture().sync();
            } catch (final InterruptedException ignored) {
            }
        }
    }
}