You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by di...@apache.org on 2022/09/06 06:42:31 UTC
[rocketmq-spring] branch master updated: [ISSUE #479] Allow programmatic modify @RocketMQMessageListener annota… (#480)
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-spring.git
The following commit(s) were added to refs/heads/master by this push:
new 607a0ef [ISSUE #479] Allow programmatic modify @RocketMQMessageListener annota… (#480)
607a0ef is described below
commit 607a0ef56c97309ffb5a54bc66d835e7007b86ee
Author: 小宇宙 <ko...@hotmail.com>
AuthorDate: Tue Sep 6 14:42:26 2022 +0800
[ISSUE #479] Allow programmatic modify @RocketMQMessageListener annota… (#480)
* [ISSUE #479]Allow programmatic modify @RocketMQMessageListener annotation properties
* add unit test
* add unit test with License
---
.../RocketMQMessageListenerBeanPostProcessor.java | 103 +++++++++++++++++++++
.../ListenerContainerConfiguration.java | 19 +---
.../autoconfigure/RocketMQAutoConfiguration.java | 4 +-
.../RocketMQListenerConfiguration.java | 38 ++++++++
...cketMQMessageListenerBeanPostProcessorTest.java | 84 +++++++++++++++++
5 files changed, 229 insertions(+), 19 deletions(-)
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListenerBeanPostProcessor.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListenerBeanPostProcessor.java
new file mode 100644
index 0000000..8226983
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListenerBeanPostProcessor.java
@@ -0,0 +1,103 @@
+/*
+ * 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.annotation;
+
+import org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.OrderComparator;
+import org.springframework.core.annotation.AnnotationUtils;
+
+import java.lang.reflect.AnnotatedElement;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.stream.Collectors;
+
+public class RocketMQMessageListenerBeanPostProcessor implements ApplicationContextAware, BeanPostProcessor, InitializingBean {
+
+ private ApplicationContext applicationContext;
+
+ private AnnotationEnhancer enhancer;
+
+ private ListenerContainerConfiguration listenerContainerConfiguration;
+
+ @Override
+ public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+ return bean;
+ }
+
+ @Override
+ public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+ Class<?> targetClass = AopUtils.getTargetClass(bean);
+ RocketMQMessageListener ann = targetClass.getAnnotation(RocketMQMessageListener.class);
+ if (ann != null) {
+ RocketMQMessageListener enhance = enhance(targetClass, ann);
+ if (listenerContainerConfiguration != null) {
+ listenerContainerConfiguration.registerContainer(beanName, bean, enhance);
+ }
+ }
+ return bean;
+ }
+
+ @Override
+ public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+ this.applicationContext = applicationContext;
+ }
+
+ @Override
+ public void afterPropertiesSet() throws Exception {
+ buildEnhancer();
+ this.listenerContainerConfiguration = this.applicationContext.getBean(ListenerContainerConfiguration.class);
+ }
+
+ private void buildEnhancer() {
+ if (this.applicationContext != null) {
+ Map<String, AnnotationEnhancer> enhancersMap =
+ this.applicationContext.getBeansOfType(AnnotationEnhancer.class, false, false);
+ if (enhancersMap.size() > 0) {
+ List<AnnotationEnhancer> enhancers = enhancersMap.values()
+ .stream()
+ .sorted(new OrderComparator())
+ .collect(Collectors.toList());
+ this.enhancer = (attrs, element) -> {
+ Map<String, Object> newAttrs = attrs;
+ for (AnnotationEnhancer enh : enhancers) {
+ newAttrs = enh.apply(newAttrs, element);
+ }
+ return attrs;
+ };
+ }
+ }
+ }
+
+ private RocketMQMessageListener enhance(AnnotatedElement element, RocketMQMessageListener ann) {
+ if (this.enhancer == null) {
+ return ann;
+ } else {
+ return AnnotationUtils.synthesizeAnnotation(
+ this.enhancer.apply(AnnotationUtils.getAnnotationAttributes(ann), element), RocketMQMessageListener.class, null);
+ }
+ }
+
+ public interface AnnotationEnhancer extends BiFunction<Map<String, Object>, AnnotatedElement, Map<String, Object>> {
+ }
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java
index fcc135f..e476927 100644
--- a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/ListenerContainerConfiguration.java
@@ -18,9 +18,7 @@
package org.apache.rocketmq.spring.autoconfigure;
import java.util.Collections;
-import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Collectors;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
@@ -33,9 +31,7 @@ import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
-import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
@@ -46,7 +42,7 @@ import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.util.StringUtils;
@Configuration
-public class ListenerContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {
+public class ListenerContainerConfiguration implements ApplicationContextAware {
private final static Logger log = LoggerFactory.getLogger(ListenerContainerConfiguration.class);
private ConfigurableApplicationContext applicationContext;
@@ -71,16 +67,7 @@ public class ListenerContainerConfiguration implements ApplicationContextAware,
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
- @Override
- public void afterSingletonsInstantiated() {
- Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class)
- .entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-
- beans.forEach(this::registerContainer);
- }
-
- private void registerContainer(String beanName, Object bean) {
+ public void registerContainer(String beanName, Object bean, RocketMQMessageListener annotation) {
Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);
if (RocketMQListener.class.isAssignableFrom(bean.getClass()) && RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
@@ -91,8 +78,6 @@ public class ListenerContainerConfiguration implements ApplicationContextAware,
throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName() + " or " + RocketMQReplyListener.class.getName());
}
- RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
-
String consumerGroup = this.environment.resolvePlaceholders(annotation.consumerGroup());
String topic = this.environment.resolvePlaceholders(annotation.topic());
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfiguration.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfiguration.java
index b84bb04..c266167 100644
--- a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfiguration.java
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQAutoConfiguration.java
@@ -55,10 +55,10 @@ import javax.annotation.PostConstruct;
@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})
+@Import({MessageConverterConfiguration.class, ListenerContainerConfiguration.class, ExtProducerResetConfiguration.class,
+ ExtConsumerResetConfiguration.class, RocketMQTransactionConfiguration.class, RocketMQListenerConfiguration.class})
@AutoConfigureAfter({MessageConverterConfiguration.class})
@AutoConfigureBefore({RocketMQTransactionConfiguration.class})
-
public class RocketMQAutoConfiguration implements ApplicationContextAware {
private static final Logger log = LoggerFactory.getLogger(RocketMQAutoConfiguration.class);
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQListenerConfiguration.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQListenerConfiguration.java
new file mode 100644
index 0000000..0761c08
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/autoconfigure/RocketMQListenerConfiguration.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.annotation.RocketMQMessageListenerBeanPostProcessor;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
+import org.springframework.core.type.AnnotationMetadata;
+
+@Configuration
+@AutoConfigureAfter(RocketMQAutoConfiguration.class)
+public class RocketMQListenerConfiguration implements ImportBeanDefinitionRegistrar {
+
+ @Override
+ public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
+ if (!registry.containsBeanDefinition(RocketMQMessageListenerBeanPostProcessor.class.getName())) {
+ registry.registerBeanDefinition(RocketMQMessageListenerBeanPostProcessor.class.getName(),
+ new RootBeanDefinition(RocketMQMessageListenerBeanPostProcessor.class));
+ }
+ }
+}
diff --git a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListenerBeanPostProcessorTest.java b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListenerBeanPostProcessorTest.java
new file mode 100644
index 0000000..58b4c6a
--- /dev/null
+++ b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListenerBeanPostProcessorTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.annotation;
+
+import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.junit.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+public class RocketMQMessageListenerBeanPostProcessorTest {
+
+ private static final String TEST_CLASS_SIMPLE_NAME = "Receiver";
+
+ private ApplicationContextRunner runner = new ApplicationContextRunner()
+ .withConfiguration(AutoConfigurations.of(RocketMQAutoConfiguration.class));
+
+ @Test
+ public void testAnnotationEnhancer() {
+ runner.withPropertyValues("rocketmq.name-server=127.0.0.1:9876").
+ withUserConfiguration(TestAnnotationEnhancerConfig.class, TestReceiverConfig.class).
+ run((context) -> {
+ // Started container failed. DefaultRocketMQListenerContainer{consumerGroup='Receiver-Custom-Consumer-Group' **
+ assertThat(context).getFailure().hasMessageContaining("connect to [127.0.0.1:9876] failed");
+ });
+
+ }
+
+ @Configuration
+ static class TestAnnotationEnhancerConfig {
+ @Bean
+ public RocketMQMessageListenerBeanPostProcessor.AnnotationEnhancer consumeContainerEnhancer() {
+ return (attrs, element) -> {
+ if (element instanceof Class) {
+ Class targetClass = (Class) element;
+ String classSimpleName = targetClass.getSimpleName();
+ if (TEST_CLASS_SIMPLE_NAME.equals(classSimpleName)) {
+ String consumerGroup = "Receiver-Custom-Consumer-Group";
+ attrs.put("consumerGroup", consumerGroup);
+ }
+ }
+ return attrs;
+ };
+ }
+ }
+
+ @Configuration
+ static class TestReceiverConfig {
+ @Bean
+ public Object receiverListener() {
+ return new Receiver();
+ }
+ }
+
+ @RocketMQMessageListener(consumerGroup = "", topic = "test")
+ static class Receiver implements RocketMQListener {
+
+ @Override
+ public void onMessage(Object message) {
+
+ }
+ }
+}