@@ -0,0 +1,210 @@
package com.njcn.message.mq ;
import com.alibaba.fastjson.JSON ;
import com.njcn.message.message.RecallMessage ;
import com.njcn.message.messagedto.MessageDataDTO ;
import com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration ;
import com.njcn.mq.autoconfig.MqCoreAutoConfiguration ;
import com.njcn.mq.container.MqListenerRegistry ;
import com.njcn.mq.core.MqTemplate ;
import com.njcn.mq.driver.redis.RedisStreamMqDriver ;
import com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration ;
import com.njcn.mq.driver.rocketmq.RocketMqDriver ;
import com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration ;
import com.njcn.mq.spi.MqDriver ;
import com.njcn.stat.api.MessAnalysisFeignClient ;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties ;
import org.apache.rocketmq.spring.core.RocketMQTemplate ;
import org.junit.jupiter.api.Assumptions ;
import org.junit.jupiter.api.Test ;
import org.springframework.boot.autoconfigure.AutoConfigurations ;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration ;
import org.springframework.boot.test.context.runner.ApplicationContextRunner ;
import org.springframework.data.redis.connection.stream.MapRecord ;
import org.springframework.data.redis.connection.stream.ReadOffset ;
import org.springframework.data.redis.connection.stream.StreamOffset ;
import org.springframework.data.redis.connection.stream.StreamReadOptions ;
import org.springframework.data.redis.connection.stream.StreamRecords ;
import org.springframework.data.redis.core.StringRedisTemplate ;
import java.net.InetSocketAddress ;
import java.net.Socket ;
import java.util.HashMap ;
import java.util.List ;
import java.util.Map ;
import static org.assertj.core.api.Assertions.assertThat ;
import static org.mockito.ArgumentMatchers.anyList ;
import static org.mockito.Mockito.mock ;
import static org.mockito.Mockito.timeout ;
import static org.mockito.Mockito.verify ;
/**
* MQ starter 切片接入验证测试（ ， ）
* 真实 Redis 要求： （ ） ；
* 纯装配用例不需要（
*
* <p>装配一律用 {@code withConfiguration(AutoConfigurations.of(...))}，让 Spring 按
* <b>真实 autoconfig 顺序</b>（ ） ，
* 验证 {@link MqCoreAutoConfiguration} 的 {@code @ConditionalOnBean(MqDriver)} 能在
* 真实顺序下拿到 driver bean —— 这依赖 driver autoconfig 上的
* {@code @AutoConfigureBefore(MqCoreAutoConfiguration.class)}。若缺该保障，按类名字母序
* {@code com.njcn.mq.autoconfig.*} 先于 {@code com.njcn.mq.driver.*}，
* bean 注册前求值条件失败，导致 MqTemplate / MqListenerRegistry 不装配。
*/
public class MqSliceWiringTest {
private static boolean redisAvailable ( ) {
try {
Socket s = new Socket ( ) ;
s . connect ( new InetSocketAddress ( " localhost " , 6379 ) , 500 ) ;
s . close ( ) ;
return true ;
} catch ( Exception e ) {
return false ;
}
}
/** redis-stream 模式：真实 autoconfig 顺序装 5 个 autoconfig + 业务 bean（
private ApplicationContextRunner redisStreamRunner ( ) {
return new ApplicationContextRunner ( )
. withConfiguration ( AutoConfigurations . of (
RedisAutoConfiguration . class ,
RedisStreamAutoConfiguration . class ,
RedisStreamMqDriverAutoConfiguration . class ,
RocketMqDriverAutoConfiguration . class ,
MqCoreAutoConfiguration . class ) )
. withUserConfiguration ( FrontDataMqListener . class , RecallMqSender . class )
. withBean ( MessAnalysisFeignClient . class ,
( ) - > mock ( MessAnalysisFeignClient . class ) )
. withPropertyValues (
" mq.type=redis-stream " ,
" spring.redis.host=localhost " ,
" spring.redis.port=6379 " ) ;
}
// ------------------------------------------------------------------
// 1. 装配：
// ------------------------------------------------------------------
@Test
void testRedisStreamDriverSelected ( ) {
// 纯装配检查，不依赖真实 Redis（
new ApplicationContextRunner ( )
. withConfiguration ( AutoConfigurations . of (
RedisAutoConfiguration . class ,
RedisStreamAutoConfiguration . class ,
RedisStreamMqDriverAutoConfiguration . class ,
RocketMqDriverAutoConfiguration . class ,
MqCoreAutoConfiguration . class ) )
. withPropertyValues (
" mq.type=redis-stream " ,
" spring.redis.host=localhost " ,
" spring.redis.port=6379 " )
. run ( ctx - > {
assertThat ( ctx ) . hasSingleBean ( MqDriver . class ) ;
assertThat ( ctx . getBean ( MqDriver . class ) ) . isInstanceOf ( RedisStreamMqDriver . class ) ;
// 真实顺序下 core 的 @ConditionalOnBean(MqDriver) 必须命中
assertThat ( ctx ) . hasSingleBean ( MqTemplate . class ) ;
assertThat ( ctx ) . hasSingleBean ( MqListenerRegistry . class ) ;
} ) ;
}
@Test
void testRocketMqDriverSelected ( ) {
// rocket 测试不依赖真实 Redis； ，
new ApplicationContextRunner ( )
. withConfiguration ( AutoConfigurations . of (
RedisStreamMqDriverAutoConfiguration . class ,
RocketMqDriverAutoConfiguration . class ,
MqCoreAutoConfiguration . class ) )
. withBean ( RocketMQTemplate . class , ( ) - > mock ( RocketMQTemplate . class ) )
. withBean ( RocketMQProperties . class , RocketMQProperties : : new )
. withPropertyValues ( " mq.type=rocketmq " )
. run ( ctx - > {
assertThat ( ctx ) . hasSingleBean ( MqDriver . class ) ;
assertThat ( ctx . getBean ( MqDriver . class ) ) . isInstanceOf ( RocketMqDriver . class ) ;
assertThat ( ctx ) . hasSingleBean ( MqTemplate . class ) ;
assertThat ( ctx ) . hasSingleBean ( MqListenerRegistry . class ) ;
} ) ;
}
// ------------------------------------------------------------------
// 2. 上行（需要真实 Redis）
// 向 LN_Topic stream XADD 一条消息，验证 FrontDataMqListener 触发了
// messAnalysisFeignClient.analysis(...)
// ------------------------------------------------------------------
@Test
void testUpstreamFrontData ( ) {
Assumptions . assumeTrue ( redisAvailable ( ) , " Redis not available at localhost:6379, skipping " ) ;
redisStreamRunner ( ) . run ( ctx - > {
StringRedisTemplate redis = ctx . getBean ( StringRedisTemplate . class ) ;
MessAnalysisFeignClient mockClient = ctx . getBean ( MessAnalysisFeignClient . class ) ;
// 构造 RedisStreamMqDriver 消费时期望的字段格式
// enc=plain → body 即为原始 JSON（ ）
// stream key = StreamKeyBuilder.buildStream("LN_Topic") = "LN_Topic"
// （默认 envIsolation=false，
MessageDataDTO dto = new MessageDataDTO ( ) ;
dto . setDataType ( 1 ) ;
String json = JSON . toJSONString ( dto ) ;
Map < String , String > fields = new HashMap < String , String > ( ) ;
fields . put ( " key " , " test-key-upstream-1 " ) ;
fields . put ( " tag " , " " ) ;
fields . put ( " enc " , " plain " ) ;
fields . put ( " body " , json ) ;
fields . put ( " ts " , String . valueOf ( System . currentTimeMillis ( ) ) ) ;
// XADD 到 LN_Topic（ ）
redis . opsForStream ( ) . add (
StreamRecords . newRecord ( ) . in ( " LN_Topic " ) . ofMap ( fields )
) ;
// 等待消费线程处理（ ，
verify ( mockClient , timeout ( 5000 ) ) . analysis ( anyList ( ) ) ;
} ) ;
}
// ------------------------------------------------------------------
// 3. 下行（需要真实 Redis）
// 通过 RecallMqSender 发送 RecallMessage，
// ------------------------------------------------------------------
@Test
void testDownstreamRecall ( ) {
Assumptions . assumeTrue ( redisAvailable ( ) , " Redis not available at localhost:6379, skipping " ) ;
redisStreamRunner ( ) . run ( ctx - > {
StringRedisTemplate redis = ctx . getBean ( StringRedisTemplate . class ) ;
RecallMqSender sender = ctx . getBean ( RecallMqSender . class ) ;
// 清理，确保只有本次发送的消息
redis . delete ( " n1_recall_Topic " ) ;
RecallMessage msg = new RecallMessage ( ) ;
msg . setGuid ( " g1 " ) ;
// MqTemplate.send("n1_recall_Topic", msg) → RedisStreamMqDriver XADD
// stream key = StreamKeyBuilder.buildStream("n1_recall_Topic") = "n1_recall_Topic"
sender . send ( msg , " n1 " ) ;
Long size = redis . opsForStream ( ) . size ( " n1_recall_Topic " ) ;
assertThat ( size ) . isGreaterThanOrEqualTo ( 1L ) ;
@SuppressWarnings ( " unchecked " )
List < MapRecord < String , Object , Object > > records =
redis . opsForStream ( ) . read (
StreamReadOptions . empty ( ) . count ( 1L ) ,
StreamOffset . create ( " n1_recall_Topic " , ReadOffset . from ( " 0-0 " ) )
) ;
assertThat ( records ) . isNotEmpty ( ) ;
Map < Object , Object > firstFields = records . get ( 0 ) . getValue ( ) ;
// DefaultMqTemplate.send(topic, payload) 使用 encodeBody(json, false) → enc="plain"
// 因此 body 即为原始 JSON，
String body = ( String ) firstFields . get ( " body " ) ;
RecallMessage result = JSON . parseObject ( body , RecallMessage . class ) ;
assertThat ( result . getGuid ( ) ) . isEqualTo ( " g1 " ) ;
} ) ;
}
}
