| 提交 | 用户 | 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 | ||
| 23 | public static void registerListenerContainer( | |
| 24 | ConfigurableListableBeanFactory beanFactory, CanalAutoConfigurationProperties canalConfig, | |
| 25 | Set<CanalListenerEndpointRegistrar> registrars) { | |
| 26 | if (registrars == null || registrars.isEmpty()) return; | |
| 27 | if (canalConfig == null || canalConfig.getInstances().isEmpty()) return; | |
| 28 | ||
| 29 | for (Map.Entry<String, CanalAutoConfigurationProperties.EndpointInstance> endpointInstance : canalConfig.getInstances().entrySet()) { | |
| 30 | if (beanFactory.containsBean(getContainerID(endpointInstance.getKey()))) continue; // 如果已经存在则不在创建 | |
| 31 | List<CanalListenerEndpointRegistrar> registrarList = new ArrayList<>(); | |
| 32 | for (CanalListenerEndpointRegistrar registrar : registrars) { | |
| 33 | if (!registrar.isContainDestination(endpointInstance.getKey())) continue; | |
| 34 | registrar.checkParameter(canalConfig.getInstances().keySet()); | |
| 35 | registrarList.add(registrar); | |
| 36 | } | |
| 37 | if (registrarList.isEmpty()) continue; | |
| 38 | registerTransponderContainer( | |
| 39 | endpointInstance.getKey(), endpointInstance.getValue(), beanFactory, registrarList); | |
| 40 | } | |
| 41 | ||
| 42 | } | |
| 43 | ||
| 44 | private static void registerTransponderContainer( | |
| 45 | String destination, CanalAutoConfigurationProperties.EndpointInstance endpointInstance, | |
| 46 | ConfigurableListableBeanFactory beanFactory, List<CanalListenerEndpointRegistrar> registrarList) { | |
| 47 | CanalConnector connector = CanalConnectorFactory.createConnector(destination, endpointInstance); | |
| 48 | beanFactory.registerSingleton(getContainerID(destination), | |
| 49 | new DmlMessageTransponderContainer(connector, registrarList, endpointInstance)); | |
| 50 | } | |
| 51 | ||
| 52 | ||
| 53 | private static String getContainerID(String destination) { | |
| 54 | return CONTAINER_ID_PREFIX + "#" + destination; | |
| 55 | } | |
| 56 | ||
| 57 | ||
| 58 | } | |
