feat(mq-starter-core): MqListenerRegistry 端点→driver.subscribe + payload/MqContext 调用

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java

@@ -0,0 +1,51 @@
+package com.njcn.mq.container;
+import com.njcn.mq.annotation.MqListener;
+import com.njcn.mq.core.MqContext;
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqMessageListener;
+import com.njcn.mq.spi.MqSubscription;
+import org.springframework.context.SmartLifecycle;
+import java.util.List;
+
+public class MqListenerRegistry implements SmartLifecycle {
+    private final MqDriver driver;
+    private final List<MqListenerEndpoint> endpoints;
+    private volatile boolean running = false;
+
+    public MqListenerRegistry(MqDriver driver, List<MqListenerEndpoint> endpoints) {
+        this.driver = driver; this.endpoints = endpoints;
+    }
+
+    @Override public synchronized void start() {
+        if (running) return;
+        for (MqListenerEndpoint ep : endpoints) {
+            MqListener meta = ep.getMeta();
+            MqMessageListener listener = (MqMessage msg) -> invoke(ep, msg);
+            MqSubscription sub = MqSubscription.builder()
+                    .topic(meta.topic()).group(meta.group()).tag(meta.tag())
+                    .payloadType(ep.getPayloadType())
+                    .concurrency(meta.concurrency()).batchSize(meta.batchSize()).maxRetry(meta.maxRetry())
+                    .listener(listener).build();
+            driver.subscribe(sub);
+        }
+        running = true;
+    }
+
+    private void invoke(MqListenerEndpoint ep, MqMessage msg) throws Exception {
+        Object[] args = ep.isWantsContext()
+                ? new Object[]{ msg.getPayload(), new MqContext(msg) }
+                : new Object[]{ msg.getPayload() };
+        try { ep.getMethod().invoke(ep.getBean(), args); }
+        catch (java.lang.reflect.InvocationTargetException e) {
+            Throwable c = e.getCause();
+            if (c instanceof Exception) throw (Exception) c;
+            throw new RuntimeException(c);
+        }
+    }
+
+    @Override public synchronized void stop() { running = false; driver.stop(); }
+    @Override public boolean isRunning() { return running; }
+    @Override public boolean isAutoStartup() { return true; }
+    @Override public int getPhase() { return Integer.MAX_VALUE - 100; }
+}

mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java

@@ -0,0 +1,38 @@
+package com.njcn.mq.container;
+import com.njcn.mq.annotation.MqListener;
+import com.njcn.mq.core.MqContext;
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqSubscription;
+import org.junit.jupiter.api.Test;
+import java.util.concurrent.atomic.AtomicReference;
+import static org.junit.jupiter.api.Assertions.*;
+
+class MqListenerRegistryTest {
+    static class Biz {
+        final AtomicReference<String> got = new AtomicReference<>();
+        final AtomicReference<String> ctxKey = new AtomicReference<>();
+        @MqListener(topic = "LN_Topic", group = "ln_consumer")
+        public void on(String payload, MqContext ctx) { got.set(payload); ctxKey.set(ctx.getKey()); }
+    }
+    /** fake driver：subscribe 时立刻把一条消息喂给 listener。 */
+    static class ImmediateDriver implements MqDriver {
+        @Override public void send(String topic, MqMessage m) {}
+        @Override public void stop() {}
+        @Override public void subscribe(MqSubscription sub) {
+            try { sub.getListener().onMessage(MqMessage.of("k9", "t", "hi")); }
+            catch (Exception e) { throw new RuntimeException(e); }
+        }
+    }
+
+    @Test
+    void start_subscribesAndInvokesBusinessMethodWithPayloadAndContext() throws Exception {
+        Biz biz = new Biz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        MqListenerRegistry reg = new MqListenerRegistry(new ImmediateDriver(), bpp.getEndpoints());
+        reg.start();
+        assertEquals("hi", biz.got.get());
+        assertEquals("k9", biz.ctxKey.get());
+    }
+}