独行的浪子
2022-04-25 3271cea1b986ee24c5bfaf2b15326554fb56090c
提交 | 用户 | age
de8c2b 1 package com.duxinglangzi.canal.starter.factory;
D 2
3 import com.alibaba.otter.canal.client.CanalConnector;
4 import com.duxinglangzi.canal.starter.configuration.CanalAutoConfigurationProperties;
5 import com.duxinglangzi.canal.starter.configuration.CanalListenerEndpointRegistrar;
6 import com.duxinglangzi.canal.starter.container.DmlMessageTransponderContainer;
7 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
8
9 import java.util.ArrayList;
10 import java.util.List;
11 import java.util.Map;
12 import java.util.Set;
13
14 /**
15  * @author wuqiong 2022/4/18
16  * @description
17  */
18 public class TransponderContainerFactory {
19
20     private static final String CONTAINER_ID_PREFIX = "com.duxinglangzi.canal.starter.container.MessageTransponderContainer#";
21
22
627420 23     /**
D 24      * 将所有待注册的端点,注册到spring中
25      * @param beanFactory
26      * @param canalConfig
27      * @param registrars
28      * @return void
29      * @author wuqiong 2022-04-23 20:34
30      */
de8c2b 31     public static void registerListenerContainer(
D 32             ConfigurableListableBeanFactory beanFactory, CanalAutoConfigurationProperties canalConfig,
33             Set<CanalListenerEndpointRegistrar> registrars) {
34         if (registrars == null || registrars.isEmpty()) return;
35         if (canalConfig == null || canalConfig.getInstances().isEmpty()) return;
36
37         for (Map.Entry<String, CanalAutoConfigurationProperties.EndpointInstance> endpointInstance : canalConfig.getInstances().entrySet()) {
38             if (beanFactory.containsBean(getContainerID(endpointInstance.getKey()))) continue; // 如果已经存在则不在创建
39             List<CanalListenerEndpointRegistrar> registrarList = new ArrayList<>();
40             for (CanalListenerEndpointRegistrar registrar : registrars) {
41                 registrar.checkParameter(canalConfig.getInstances().keySet());
627420 42                 if (!registrar.isContainDestination(endpointInstance.getKey())) continue;
de8c2b 43                 registrarList.add(registrar);
D 44             }
45             if (registrarList.isEmpty()) continue;
46             registerTransponderContainer(
47                     endpointInstance.getKey(), endpointInstance.getValue(), beanFactory, registrarList);
48         }
49
50     }
51
52     private static void registerTransponderContainer(
53             String destination, CanalAutoConfigurationProperties.EndpointInstance endpointInstance,
54             ConfigurableListableBeanFactory beanFactory, List<CanalListenerEndpointRegistrar> registrarList) {
55         CanalConnector connector = CanalConnectorFactory.createConnector(destination, endpointInstance);
56         beanFactory.registerSingleton(getContainerID(destination),
57                 new DmlMessageTransponderContainer(connector, registrarList, endpointInstance));
58     }
59
60
61     private static String getContainerID(String destination) {
62         return CONTAINER_ID_PREFIX + "#" + destination;
63     }
64
65
66 }