[ISSUE #632 ] Fix NPE caused by using ExtRocketMQTemplateConfiguratio…

…n annotation extension to send messages
apache · Mar 3, 2024 · b03f552 · b03f552
1 parent 404085e
commit b03f552
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 9 deletions.
diff --git a/.../java/org/apache/rocketmq/spring/annotation/RocketMQMessageListenerBeanPostProcessor.java b/.../java/org/apache/rocketmq/spring/annotation/RocketMQMessageListenerBeanPostProcessor.java
@@ -23,6 +23,7 @@
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.SmartLifecycle;
 import org.springframework.core.OrderComparator;
 import org.springframework.core.annotation.AnnotationUtils;
 
@@ -32,14 +33,16 @@
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 
-public class RocketMQMessageListenerBeanPostProcessor implements ApplicationContextAware, BeanPostProcessor, InitializingBean {
+public class RocketMQMessageListenerBeanPostProcessor implements ApplicationContextAware, BeanPostProcessor, InitializingBean, SmartLifecycle {
 
     private ApplicationContext applicationContext;
 
     private AnnotationEnhancer enhancer;
 
     private RocketMQMessageListenerContainerRegistrar listenerContainerRegistrar;
 
+    private boolean running = false;
+
     @Override
     public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
         return bean;
@@ -58,6 +61,34 @@ public Object postProcessAfterInitialization(Object bean, String beanName) throw
         return bean;
     }
 
+    @Override
+    public int getPhase() {
+        return Integer.MAX_VALUE - 2000;
+    }
+
+    @Override
+    public void start() {
+        if (!isRunning()) {
+            this.setRunning(true);
+            listenerContainerRegistrar.startContainer();
+        }
+    }
+
+    @Override
+    public void stop() {
+
+    }
+
+    public void setRunning(boolean running) {
+        this.running = running;
+    }
+
+
+    @Override
+    public boolean isRunning() {
+        return running;
+    }
+
     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
         this.applicationContext = applicationContext;

diff --git a/...in/java/org/apache/rocketmq/spring/support/RocketMQMessageListenerContainerRegistrar.java b/...in/java/org/apache/rocketmq/spring/support/RocketMQMessageListenerContainerRegistrar.java
@@ -36,7 +36,9 @@
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.util.StringUtils;
 
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicLong;
 
 public class RocketMQMessageListenerContainerRegistrar implements ApplicationContextAware {
@@ -52,6 +54,8 @@ public class RocketMQMessageListenerContainerRegistrar implements ApplicationCon
 
     private final RocketMQMessageConverter rocketMQMessageConverter;
 
+    private final List<DefaultRocketMQListenerContainer> containers = new ArrayList<>();
+
     public RocketMQMessageListenerContainerRegistrar(RocketMQMessageConverter rocketMQMessageConverter,
                                                      ConfigurableEnvironment environment, RocketMQProperties rocketMQProperties) {
         this.rocketMQMessageConverter = rocketMQMessageConverter;
@@ -97,18 +101,25 @@ public void registerContainer(String beanName, Object bean, RocketMQMessageListe
         genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class, () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
         DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,
                 DefaultRocketMQListenerContainer.class);
-        if (!container.isRunning()) {
-            try {
-                container.start();
-            } catch (Exception e) {
-                log.error("Started container failed. {}", container, e);
-                throw new RuntimeException(e);
-            }
-        }
+
+        containers.add(container);
 
         log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
     }
 
+    public void startContainer() {
+        for (DefaultRocketMQListenerContainer container : containers) {
+            if (!container.isRunning()) {
+                try {
+                    container.start();
+                } catch (Exception e) {
+                    log.error("Started container failed. {}", container, e);
+                    throw new RuntimeException(e);
+                }
+            }
+        }
+    }
+
     private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean,
                                                                              RocketMQMessageListener annotation) {
         DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();