diff --git a/mq-starter-rocketmq/pom.xml b/mq-starter-rocketmq/pom.xml index d95a801..500da39 100644 --- a/mq-starter-rocketmq/pom.xml +++ b/mq-starter-rocketmq/pom.xml @@ -3,4 +3,12 @@ 4.0.0 com.njcnmq-starter1.0.0-SNAPSHOT mq-starter-rocketmq + + com.njcnmq-starter-core1.0.0-SNAPSHOT + org.apache.rocketmqrocketmq-spring-boot-starter2.2.3 + org.springframework.bootspring-boot-autoconfigure${autoconfigure.version} + com.alibabafastjson${fastjson.version} + org.projectlomboklombok${lombok.version}true + org.springframework.bootspring-boot-starter-test${springboot.version}test + diff --git a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java new file mode 100644 index 0000000..3a4b43e --- /dev/null +++ b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java @@ -0,0 +1,73 @@ +package com.njcn.mq.driver.rocketmq; + +import com.alibaba.fastjson.JSON; +import com.njcn.mq.core.MqMessage; +import com.njcn.mq.spi.MqDriver; +import com.njcn.mq.spi.MqSubscription; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.apache.rocketmq.spring.support.RocketMQHeaders; +import org.springframework.messaging.support.MessageBuilder; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; + +@Slf4j +public class RocketMqDriver implements MqDriver { + + private final RocketMQTemplate template; + private final String namesrvAddr; + private final List consumers = new ArrayList<>(); + + public RocketMqDriver(RocketMQTemplate template, String namesrvAddr) { + this.template = template; + this.namesrvAddr = namesrvAddr; + } + + @Override + public void send(String topic, MqMessage message) { + String tag = message.getTag(); + String dest = (tag == null || tag.isEmpty()) ? topic : topic + ":" + tag; + String json = JSON.toJSONString(message.getPayload()); + template.syncSend(dest, MessageBuilder.withPayload(json) + .setHeader(RocketMQHeaders.KEYS, message.getKey()).build()); + } + + @Override + public void subscribe(MqSubscription sub) { + try { + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(sub.getGroup()); + consumer.setNamesrvAddr(namesrvAddr); + String tag = sub.getTag(); + consumer.subscribe(sub.getTopic(), (tag == null || tag.isEmpty()) ? "*" : tag); + consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { + for (MessageExt ext : msgs) { + try { + String json = new String(ext.getBody(), StandardCharsets.UTF_8); + Object payload = JSON.parseObject(json, sub.getPayloadType()); + sub.getListener().onMessage(MqMessage.of(ext.getKeys(), ext.getTags(), payload)); + } catch (Exception e) { + log.warn("[mq-rocket] handle failed msgId={}", ext.getMsgId(), e); + return ConsumeConcurrentlyStatus.RECONSUME_LATER; + } + } + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + }); + consumer.start(); + consumers.add(consumer); + } catch (Exception e) { + throw new IllegalStateException("rocket subscribe failed: " + sub.getTopic(), e); + } + } + + @Override + public void stop() { + for (DefaultMQPushConsumer c : consumers) c.shutdown(); + consumers.clear(); + } +} diff --git a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriverAutoConfiguration.java b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriverAutoConfiguration.java new file mode 100644 index 0000000..b7a9093 --- /dev/null +++ b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriverAutoConfiguration.java @@ -0,0 +1,22 @@ +package com.njcn.mq.driver.rocketmq; + +import com.njcn.mq.spi.MqDriver; +import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +@ConditionalOnClass(RocketMQTemplate.class) +@ConditionalOnProperty(name = "mq.type", havingValue = "rocketmq", matchIfMissing = true) +public class RocketMqDriverAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public MqDriver mqDriver(RocketMQTemplate template, RocketMQProperties props) { + return new RocketMqDriver(template, props.getNameServer()); + } +} diff --git a/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories b/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000..e317b90 --- /dev/null +++ b/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration diff --git a/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverSendTest.java b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverSendTest.java new file mode 100644 index 0000000..da46fca --- /dev/null +++ b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverSendTest.java @@ -0,0 +1,31 @@ +package com.njcn.mq.driver.rocketmq; + +import com.njcn.mq.core.MqMessage; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.messaging.Message; + +import static org.mockito.Mockito.*; +import static org.junit.jupiter.api.Assertions.*; + +class RocketMqDriverSendTest { + @Test + void send_withTag_buildsDestinationAndKeysHeader() { + RocketMQTemplate template = mock(RocketMQTemplate.class); + RocketMqDriver driver = new RocketMqDriver(template, "localhost:9876"); + driver.send("File_Topic", MqMessage.of("k1", "node-7", "p")); + ArgumentCaptor dest = ArgumentCaptor.forClass(String.class); + verify(template).syncSend(dest.capture(), any(Message.class)); + assertEquals("File_Topic:node-7", dest.getValue()); + } + + @Test + void send_noTag_destinationIsTopicOnly() { + RocketMQTemplate template = mock(RocketMQTemplate.class); + new RocketMqDriver(template, "localhost:9876").send("LN_Topic", MqMessage.of("k", "", "p")); + ArgumentCaptor dest = ArgumentCaptor.forClass(String.class); + verify(template).syncSend(dest.capture(), any(Message.class)); + assertEquals("LN_Topic", dest.getValue()); + } +}