[ISSUE #583]: 新增 扩展点用于实现链路追踪

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java

@@ -18,6 +18,7 @@
 package org.apache.rocketmq.spring.autoconfigure;
 
 import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
+import org.apache.rocketmq.spring.support.RocketMQMessageHandler;
 import org.apache.rocketmq.spring.support.RocketMQMessageListenerContainerRegistrar;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
@@ -28,7 +29,7 @@
 @ConditionalOnMissingBean(RocketMQMessageListenerContainerRegistrar.class)
 public class ListenerContainerConfiguration {
     @Bean
-    public RocketMQMessageListenerContainerRegistrar rocketMQMessageListenerContainerRegistrar(RocketMQMessageConverter rocketMQMessageConverter, ConfigurableEnvironment environment, RocketMQProperties rocketMQProperties) {
-        return new RocketMQMessageListenerContainerRegistrar(rocketMQMessageConverter, environment, rocketMQProperties);
+    public RocketMQMessageListenerContainerRegistrar rocketMQMessageListenerContainerRegistrar(RocketMQMessageConverter rocketMQMessageConverter, ConfigurableEnvironment environment, RocketMQProperties rocketMQProperties, RocketMQMessageHandler rocketMQMessageHandler) {
+        return new RocketMQMessageListenerContainerRegistrar(rocketMQMessageConverter, environment, rocketMQProperties, rocketMQMessageHandler);
     }
 }

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/MessagePostProcessorConfiguration.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.spring.autoconfigure;
+
+
+import org.apache.rocketmq.spring.support.RocketMQMessagePostProcessor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @see RocketMQMessagePostProcessor
+ */
+@Configuration
+@ConditionalOnMissingBean(RocketMQMessagePostProcessor.class)
+class MessagePostProcessorConfiguration {
+
+    @Bean
+    public RocketMQMessagePostProcessor createMessagePostProcessor() {
+        return new RocketMQMessagePostProcessor();
+    }
+
+}

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfiguration.java

@@ -25,6 +25,7 @@
 import org.apache.rocketmq.spring.annotation.SelectorType;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
+import org.apache.rocketmq.spring.support.RocketMQMessagePostProcessor;
 import org.apache.rocketmq.spring.support.RocketMQUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -51,9 +52,10 @@
 @EnableConfigurationProperties(RocketMQProperties.class)
 @ConditionalOnClass({MQAdmin.class})
 @ConditionalOnProperty(prefix = "rocketmq", value = "name-server", matchIfMissing = true)
-@Import({MessageConverterConfiguration.class, ListenerContainerConfiguration.class, ExtProducerResetConfiguration.class,
-        ExtConsumerResetConfiguration.class, RocketMQTransactionConfiguration.class, RocketMQListenerConfiguration.class})
-@AutoConfigureAfter({MessageConverterConfiguration.class})
+@Import({MessageConverterConfiguration.class, RocketMQMessageHandlerConfiguration.class, ListenerContainerConfiguration.class, ExtProducerResetConfiguration.class,
+        ExtConsumerResetConfiguration.class, RocketMQTransactionConfiguration.class, RocketMQListenerConfiguration.class,
+        MessagePostProcessorConfiguration.class})
+@AutoConfigureAfter({MessageConverterConfiguration.class, MessagePostProcessorConfiguration.class, RocketMQMessageHandlerConfiguration.class})
 @AutoConfigureBefore({RocketMQTransactionConfiguration.class})
 public class RocketMQAutoConfiguration implements ApplicationContextAware {
     private static final Logger log = LoggerFactory.getLogger(RocketMQAutoConfiguration.class);
@@ -166,7 +168,8 @@ public DefaultLitePullConsumer defaultLitePullConsumer(RocketMQProperties rocket
     @Bean(destroyMethod = "destroy")
     @Conditional(ProducerOrConsumerPropertyCondition.class)
     @ConditionalOnMissingBean(name = ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME)
-    public RocketMQTemplate rocketMQTemplate(RocketMQMessageConverter rocketMQMessageConverter) {
+    public RocketMQTemplate rocketMQTemplate(RocketMQMessageConverter rocketMQMessageConverter,
+                                             RocketMQMessagePostProcessor rocketMQMessagePostProcessor) {
         RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
         if (applicationContext.containsBean(PRODUCER_BEAN_NAME)) {
             rocketMQTemplate.setProducer((DefaultMQProducer) applicationContext.getBean(PRODUCER_BEAN_NAME));
@@ -175,6 +178,7 @@ public RocketMQTemplate rocketMQTemplate(RocketMQMessageConverter rocketMQMessag
             rocketMQTemplate.setConsumer((DefaultLitePullConsumer) applicationContext.getBean(CONSUMER_BEAN_NAME));
         }
         rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
+        rocketMQTemplate.setMessagePostProcessor(rocketMQMessagePostProcessor.getMessagePostProcessor());
         return rocketMQTemplate;
     }
 

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQMessageHandlerConfiguration.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.spring.autoconfigure;
+
+import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
+import org.apache.rocketmq.spring.support.RocketMQMessageHandler;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @see RocketMQMessageConverter
+ */
+@Configuration
+@ConditionalOnMissingBean(RocketMQMessageHandler.class)
+class RocketMQMessageHandlerConfiguration {
+
+    @Bean
+    public RocketMQMessageHandler createRocketMQMessageHandler() {
+        return (message, consumer) -> consumer.doHandler(message);
+    }
+
+}

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java

@@ -72,6 +72,8 @@ public class RocketMQTemplate extends AbstractMessageSendingTemplate<String> imp
 
     private RocketMQMessageConverter rocketMQMessageConverter = new RocketMQMessageConverter();
 
+    private MessagePostProcessor messagePostProcessor;
+
     public DefaultMQProducer getProducer() {
         return producer;
     }
@@ -104,6 +106,14 @@ public void setMessageQueueSelector(MessageQueueSelector messageQueueSelector) {
         this.messageQueueSelector = messageQueueSelector;
     }
 
+    public MessagePostProcessor getMessagePostProcessor() {
+        return messagePostProcessor;
+    }
+
+    public void setMessagePostProcessor(MessagePostProcessor messagePostProcessor) {
+        this.messagePostProcessor = messagePostProcessor;
+    }
+
     public void setAsyncSenderExecutor(ExecutorService asyncSenderExecutor) {
         this.producer.setAsyncSenderExecutor(asyncSenderExecutor);
     }
@@ -1191,7 +1201,7 @@ public TransactionSendResult sendMessageInTransaction(final String destination,
 
     private org.apache.rocketmq.common.message.Message createRocketMqMessage(
         String destination, Message<?> message) {
-        Message<?> msg = this.doConvert(message.getPayload(), message.getHeaders(), null);
+        Message<?> msg = this.doConvert(message.getPayload(), message.getHeaders(), messagePostProcessor);
         return RocketMQUtil.convertToRocketMessage(getMessageConverter(), charset,
             destination, msg);
     }

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java

@@ -140,6 +140,8 @@ public class DefaultRocketMQListenerContainer implements InitializingBean,
 
     private String instanceName;
 
+    private RocketMQMessageHandler rocketMQMessageHandler = (message, chain) -> chain.doHandler(message);
+
     public long getSuspendCurrentQueueTimeMillis() {
         return suspendCurrentQueueTimeMillis;
     }
@@ -318,6 +320,14 @@ public void setInstanceName(String instanceName) {
         this.instanceName = instanceName;
     }
 
+    public RocketMQMessageHandler getRocketMQMessageHandler() {
+        return rocketMQMessageHandler;
+    }
+
+    public void setRocketMQMessageHandler(RocketMQMessageHandler rocketMQMessageHandler) {
+        this.rocketMQMessageHandler = rocketMQMessageHandler;
+    }
+
     public DefaultRocketMQListenerContainer setAwaitTerminationMillisWhenShutdown(long awaitTerminationMillisWhenShutdown) {
         this.awaitTerminationMillisWhenShutdown = awaitTerminationMillisWhenShutdown;
         return this;
@@ -426,15 +436,21 @@ public class DefaultMessageListenerConcurrently implements MessageListenerConcur
         @Override
         public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
             for (MessageExt messageExt : msgs) {
-                log.debug("received msg: {}", messageExt);
                 try {
-                    long now = System.currentTimeMillis();
-                    DefaultRocketMQListenerContainer container = applicationContext.getBean(name, DefaultRocketMQListenerContainer.class);
-                    container.handleMessage(messageExt);
-                    long costTime = System.currentTimeMillis() - now;
-                    log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
+                    rocketMQMessageHandler.doHandler(messageExt, messageExt1 -> {
+                        log.debug("received msg: {}", messageExt1);
+                        try {
+                            long now = System.currentTimeMillis();
+                            DefaultRocketMQListenerContainer container = applicationContext.getBean(name, DefaultRocketMQListenerContainer.class);
+                            container.handleMessage(messageExt1);
+                            long costTime = System.currentTimeMillis() - now;
+                            log.debug("consume {} cost: {} ms", messageExt1.getMsgId(), costTime);
+                        } catch (Exception e) {
+                            log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt1.getMsgId(), messageExt1.getTopic(), messageExt1.getReconsumeTimes(), e);
+                            throw e;
+                        }
+                    });
                 } catch (Exception e) {
-                    log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e);
                     context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                 }
@@ -450,15 +466,21 @@ public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
         @Override
         public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
             for (MessageExt messageExt : msgs) {
-                log.debug("received msg: {}", messageExt);
                 try {
-                    long now = System.currentTimeMillis();
-                    DefaultRocketMQListenerContainer container = applicationContext.getBean(name, DefaultRocketMQListenerContainer.class);
-                    container.handleMessage(messageExt);
-                    long costTime = System.currentTimeMillis() - now;
-                    log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
+                    rocketMQMessageHandler.doHandler(messageExt, messageExt1 -> {
+                        log.debug("received msg: {}", messageExt1);
+                        try {
+                            long now = System.currentTimeMillis();
+                            DefaultRocketMQListenerContainer container = applicationContext.getBean(name, DefaultRocketMQListenerContainer.class);
+                            container.handleMessage(messageExt1);
+                            long costTime = System.currentTimeMillis() - now;
+                            log.debug("consume {} cost: {} ms", messageExt1.getMsgId(), costTime);
+                        } catch (Exception e) {
+                            log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt1.getMsgId(), messageExt1.getTopic(), messageExt1.getReconsumeTimes(), e);
+                            throw e;
+                        }
+                    });
                 } catch (Exception e) {
-                    log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e);
                     context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);
                     return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                 }

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQMessageHandler.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.spring.support;
+
+
+import org.apache.rocketmq.common.message.MessageExt;
+
+public interface RocketMQMessageHandler {
+
+    void doHandler(MessageExt message, RocketMQMessageHandlerChain chain) throws Exception;
+
+}

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQMessageHandlerChain.java

@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.spring.support;
+
+import org.apache.rocketmq.common.message.MessageExt;
+
+public interface RocketMQMessageHandlerChain {
+    void doHandler(MessageExt message) throws Exception;
+}

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQMessageListenerContainerRegistrar.java

@@ -56,11 +56,15 @@ public class RocketMQMessageListenerContainerRegistrar implements ApplicationCon
 
     private final List<DefaultRocketMQListenerContainer> containers = new ArrayList<>();
 
+    private final RocketMQMessageHandler rocketMQMessageHandler;
+
     public RocketMQMessageListenerContainerRegistrar(RocketMQMessageConverter rocketMQMessageConverter,
-                                                     ConfigurableEnvironment environment, RocketMQProperties rocketMQProperties) {
+                                                     ConfigurableEnvironment environment, RocketMQProperties rocketMQProperties,
+                                                     RocketMQMessageHandler rocketMQMessageHandler) {
         this.rocketMQMessageConverter = rocketMQMessageConverter;
         this.environment = environment;
         this.rocketMQProperties = rocketMQProperties;
+        this.rocketMQMessageHandler = rocketMQMessageHandler;
     }
 
     @Override
@@ -146,6 +150,7 @@ private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String
             container.setRocketMQReplyListener((RocketMQReplyListener) bean);
         }
         container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
+        container.setRocketMQMessageHandler(rocketMQMessageHandler);
         container.setName(name);
 
         String namespace = environment.resolvePlaceholders(annotation.namespace());

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQMessagePostProcessor.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.spring.support;
+
+import org.springframework.messaging.core.MessagePostProcessor;
+
+/**
+ * @see MessagePostProcessor
+ */
+public class RocketMQMessagePostProcessor {
+
+    public MessagePostProcessor getMessagePostProcessor() {
+        return null;
+    }
+
+}