duxinglangzi
2022-04-24 fa654ac361fcc231bc389ca446f78d501c7060c8
提交 | 用户 | age
de8c2b 1 package com.duxinglangzi.canal.starter.configuration;
D 2
3
4 import com.duxinglangzi.canal.starter.annotation.CanalListener;
5 import com.duxinglangzi.canal.starter.factory.TransponderContainerFactory;
6 import org.slf4j.Logger;
7 import org.slf4j.LoggerFactory;
8 import org.springframework.aop.support.AopUtils;
9 import org.springframework.beans.BeansException;
10 import org.springframework.beans.factory.SmartInitializingSingleton;
11 import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
12 import org.springframework.beans.factory.config.BeanPostProcessor;
13 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
14 import org.springframework.core.MethodIntrospector;
15 import org.springframework.core.annotation.AnnotatedElementUtils;
16
17 import java.lang.reflect.Method;
18 import java.util.Collections;
19 import java.util.Map;
20 import java.util.Set;
21 import java.util.concurrent.ConcurrentHashMap;
22 import java.util.stream.Collectors;
23
24 /**
25  * @author wuqiong 2022/4/11
26  */
27 public class CanalListenerAnnotationBeanPostProcessor implements
28         BeanPostProcessor, SmartInitializingSingleton, BeanFactoryPostProcessor {
29
30     private static final Logger logger = LoggerFactory.getLogger(CanalListenerAnnotationBeanPostProcessor.class);
31
32     private final Set<Class<?>> notAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap<>(256));
33     private Set<CanalListenerEndpointRegistrar> registrars = Collections.newSetFromMap(new ConcurrentHashMap<>());
34     private ConfigurableListableBeanFactory configurableListableBeanFactory;
35     private CanalAutoConfigurationProperties canalAutoConfigurationProperties;
36
37     @Override
38     public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
39         if (notAnnotatedClasses.contains(bean.getClass())) return bean;
40         Class<?> targetClass = AopUtils.getTargetClass(bean);
627420 41         // 只扫描类的方法,目前 CanalListener 只支持在方法上
de8c2b 42         Map<Method, CanalListener> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
D 43                 (MethodIntrospector.MetadataLookup<CanalListener>) method -> findListenerAnnotations(method));
44         if (annotatedMethods.isEmpty()) {
45             this.notAnnotatedClasses.add(bean.getClass());
46         } else {
627420 47             // 先加入到待注册里面
de8c2b 48             annotatedMethods.entrySet().stream()
D 49                     .filter(e -> e != null)
50                     .forEach(ele -> registrars.add(new CanalListenerEndpointRegistrar(bean, ele)));
627420 51             logger.info("Registered @CanalListener methods processed on bean:{} , Methods :{} ", beanName,
de8c2b 52                     annotatedMethods.keySet().stream().map(e -> e.getName()).collect(Collectors.toSet()));
D 53         }
54         return bean;
55     }
56
57
58     private CanalListener findListenerAnnotations(Method method) {
59         return AnnotatedElementUtils.findMergedAnnotation(method, CanalListener.class);
60     }
61
62
63     @Override
64     public void afterSingletonsInstantiated() {
65         canalAutoConfigurationProperties = configurableListableBeanFactory.getBean(CanalAutoConfigurationProperties.class);
66         TransponderContainerFactory.registerListenerContainer(configurableListableBeanFactory, canalAutoConfigurationProperties, registrars);
67     }
68
69     @Override
70     public void postProcessBeanFactory(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
71         this.configurableListableBeanFactory = configurableListableBeanFactory;
72     }
73
74
75 }