package com.ruoyi.web.core.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

/**
 * @Description: canal配置类
 * @Author: huangcheng
 * @Date: 2021/8/16
 * @Version V1.0
 */
@Component
@Slf4j
public class CanalClient implements DisposableBean {

    private CanalConnector canalConnector;

    //    @Bean
    public CanalConnector getCanalConnector() {
        canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
                11111), "heiyan", "", "");
        canalConnector.connect();
        //指定filter,格式{database}.{table}
        canalConnector.subscribe();
        //回滚寻找上次中断的为止
        canalConnector.rollback();
        return canalConnector;
    }

    /**
     * 在spring容器销毁的时候，需要断开canal客户端的连接
     * 防止canal连接的泄露
     *
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        if (canalConnector != null) {
            log.info("断开canal客户端的连接");
            canalConnector.disconnect();
        }
    }
}
