publishSubscribeChannel 单元测试不能正常工作答案

【问题标题】：publishSubscribeChannel unit test can't work wellpublishSubscribeChannel 单元测试不能正常工作
【发布时间】：2018-12-28 11:54:59
【问题描述】：

我的集成配置类如下，当我对它们进行一些单元测试时，发现：当我向UserRecipientSubscribeCacheChannel 发送消息时，它运行良好；当我向频道userReportWriteCompletedRouteChannel 的上层发送消息时，它工作失败，并且它还没有抛出任何异常。我无法理解。当然，我发送的消息是相同的。

由于失败部分，下一个处理程序无法正常工作。太棒了！！

它在下面工作正常，它打印===>ip location channel message:GenericMessage [payload=[MailRecipientActionDocumen...和===>user recipient channel message:GenericMessage [payload=[UserRecipientSubscribeDataRedisStructure...

 @Test
public void test_sendMessageUserRecipientSubscribeCacheChannel(){
    UserRecipientSubscribeCacheChannel.send(createMessageWithIp());
}

它在下面工作失败，它只打印===>ip location channel message:GenericMessage [payload=[MailRecipientActionDocumen...

注意：fail部分，在handler前面有一个transformer。

@Test
public void test_sendMessageToRouteChannel() {
    userReportWriteCompletedRouteChannel.send(createMessageWithIp());
}

我的代码配置如下：

@Bean  
public SubscribableChannel userReportWriteCompletedSubscribeChannel() {  
    return new DirectChannel();  
} 

@Bean
public QueueChannel userReportWriteCompletedRouteChannel() {
    return new QueueChannel();
}

@Bean
public MessageChannel ipLocationResolveCacheChannel() {
    return new DirectChannel();
}

@Bean
public MessageChannel userRecipientSubscribeCacheChannel() {
    return new DirectChannel();
}

@MessagingGateway(name = "userReportWriteCompletedListener",
        defaultRequestChannel = "userReportWriteCompletedRouteChannel")
public interface UserReportWriteCompletedListener {
    @Gateway
    void receive(List<UserMailRecipientActionDocument> docs);
}

@Bean
public IntegrationFlow bridgeFlow() {
    return flow -> flow.channel("userReportWriteCompletedRouteChannel")
            .bridge(bridgeSpe -> bridgeSpe
                    .poller(pollerFactory -> pollerFactory.fixedRate(500).maxMessagesPerPoll(1)))

            .channel("userReportWriteCompletedSubscribeChannel")
            ;
}

@Bean
public IntegrationFlow subscribeFlow() {
    return IntegrationFlows.from("userReportWriteCompletedSubscribeChannel")
            .publishSubscribeChannel(publishSubscribeSpec -> publishSubscribeSpec
                    .subscribe(flow -> flow
                            .channel(IP_LOCATION_RESOLVE_CACHE_CHANNEL)
                    )
                    .subscribe(flow -> flow
                            .channel(USER_RECIPIENT_SUBSCRIBE_CACHE_CHANNEL)
                    ))
            .get();
}

@Bean
public RedisStoreWritingMessageHandler ipLocationResolveCacheHandler(RedisTemplate<String, ?> redisTemplate) {
    final RedisStoreWritingMessageHandler ipLocationResolveCacheHandler =
            new RedisStoreWritingMessageHandler(redisTemplate);
    ipLocationResolveCacheHandler.setKey("IP_LOCATION_RESOLVE_CACHE");
    return ipLocationResolveCacheHandler;
}

@Bean
public RedisStoreWritingMessageHandler userRecipientSubscribeCacheHandler(RedisTemplate<String, ?> redisTemplate) {
    final RedisStoreWritingMessageHandler userRecipientSubscribeCacheHandler =
            new RedisStoreWritingMessageHandler(redisTemplate);
    userRecipientSubscribeCacheHandler.setKey("USER_RECIPIENT_SUBSCRIBE_CACHE");
    return userRecipientSubscribeCacheHandler;
}

@Bean
public IpLocationResolveRedisStructureFilterAndTransformer recipientActionHasIpFilterAndTransformer() {
    return new IpLocationResolveRedisStructureFilterAndTransformer();
}

@Bean
public UserRecipientSubscribeDataRedisStructureTransformer subscribeDataRedisStructureTransformer(
        IpLocationClient ipLocationClient) {
    return new UserRecipientSubscribeDataRedisStructureTransformer(ipLocationClient);
}

@Bean
public IntegrationFlow ipLocationResolveCacheFlow(
        @Qualifier("ipLocationResolveCacheHandler") RedisStoreWritingMessageHandler writingMessageHandler) {
    return flow -> flow.channel(IP_LOCATION_RESOLVE_CACHE_CHANNEL)
            .handle(message -> {
                System.out.println("===>ip location  channel message:" + message);
            })

            ;
}

@Bean
public IntegrationFlow userRecipientActionDataCacheFlow(
        @Qualifier("userRecipientSubscribeCacheHandler") RedisStoreWritingMessageHandler messageHandler,
        UserRecipientSubscribeDataRedisStructureTransformer transformer) {
    return flow -> flow.channel(USER_RECIPIENT_SUBSCRIBE_CACHE_CHANNEL)
            .transform(transformer)
                   .handle(message -> {
                System.out.println("===>user recipient  channel message:" + message);
            })      
}

我希望打印 2 个消息信息，但只打印 1 个。

【问题讨论】：

标签： spring-integration spring-integration-dsl

【解决方案1】：

今天发现桥流可能有问题，当我将handler移到channeluserReportWriteCompletedSubscribeChannel后面时，无法打印任何消息；当我删除通道并直接添加处理程序时，它将打印消息。我是不是用错了桥接器？

   @Bean
public IntegrationFlow bridgeFlow() {
    return flow -> flow.channel("userReportWriteCompletedRouteChannel")
            .bridge(bridgeSpe -> bridgeSpe
                    .poller(pollerFactory -> pollerFactory.fixedRate(100).maxMessagesPerPoll(1)))
            .handle(message -> {
                System.out.println("===>route  channel message:" + message);
            })   // handle ok ,  will print message
            .channel("userReportWriteCompletedSubscribeChannel")
            // .handle(message -> {
            //    System.out.println("===>route  channel message:" + message);
            // })       // handle fail ,  will not printing message
            ;
}

测试：

    @Test
//invalid
public void test_sendMessageToRouteChannel() {
    userReportWriteCompletedRouteChannel.send(createMessageWithIp());
}

【讨论】：