diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java new file mode 100644 index 0000000..72d3125 --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java @@ -0,0 +1,35 @@ +package com.njcn.mq.container; +import com.njcn.mq.annotation.MqListener; +import com.njcn.mq.core.MqContext; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.core.annotation.AnnotatedElementUtils; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; + +public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor { + + private final List endpoints = new ArrayList<>(); + public List getEndpoints() { return endpoints; } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + for (Method m : bean.getClass().getDeclaredMethods()) { + MqListener meta = AnnotatedElementUtils.findMergedAnnotation(m, MqListener.class); + if (meta == null) continue; + Class[] params = m.getParameterTypes(); + if (params.length < 1 || params.length > 2) { + throw new IllegalStateException("@MqListener 方法签名须为 (payload) 或 (payload, MqContext): " + m); + } + boolean wantsContext = params.length == 2; + if (wantsContext && params[1] != MqContext.class) { + throw new IllegalStateException("@MqListener 第二参须为 MqContext: " + m); + } + Class payloadType = params[0]; + m.setAccessible(true); + endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext)); + } + return bean; + } +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerEndpoint.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerEndpoint.java new file mode 100644 index 0000000..366b562 --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerEndpoint.java @@ -0,0 +1,15 @@ +package com.njcn.mq.container; +import com.njcn.mq.annotation.MqListener; +import lombok.AllArgsConstructor; +import lombok.Data; +import java.lang.reflect.Method; + +@Data +@AllArgsConstructor +public class MqListenerEndpoint { + private Object bean; + private Method method; + private MqListener meta; + private Class payloadType; + private boolean wantsContext; +} diff --git a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java new file mode 100644 index 0000000..f65fef2 --- /dev/null +++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java @@ -0,0 +1,28 @@ +package com.njcn.mq.container; +import com.njcn.mq.annotation.MqListener; +import com.njcn.mq.core.MqContext; +import org.junit.jupiter.api.Test; +import java.util.List; +import static org.junit.jupiter.api.Assertions.*; + +class MqListenerScanTest { + static class Biz { + @MqListener(topic = "LN_Topic", group = "ln_consumer") + public void onData(String msg) { } + @MqListener(topic = "T2", group = "g2") + public void onWithCtx(String msg, MqContext ctx) { } + } + + @Test + void scan_findsEndpoints_withPayloadTypeAndContextFlag() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new Biz(), "biz"); + List eps = bpp.getEndpoints(); + assertEquals(2, eps.size()); + MqListenerEndpoint e1 = eps.stream().filter(e -> e.getMeta().topic().equals("LN_Topic")).findFirst().orElseThrow(AssertionError::new); + assertEquals(String.class, e1.getPayloadType()); + assertFalse(e1.isWantsContext()); + MqListenerEndpoint e2 = eps.stream().filter(e -> e.getMeta().topic().equals("T2")).findFirst().orElseThrow(AssertionError::new); + assertTrue(e2.isWantsContext()); + } +}