Spring Integration集成测试：断言消息已发送至指定DiscardChannel

我维护一个基于Spring Boot的事件驱动项目，使用Spring Integration库，所有通道均为DirectChannel（因为从Kafka接收的单个事件采用单线程端到端处理）。现在需要为验证失败的场景编写集成测试，核心断言是消息已成功发送至DiscardChannel。

我自己想到了两种可行方案：

方案1：替换为PublishSubscribeChannel并添加虚拟订阅者 #

通过@TestConfiguration替换原有的DiscardChannel为PublishSubscribeChannel，支持多订阅者，既保留原有业务逻辑，又能添加测试用的虚拟订阅者收集消息做断言：

@TestConfiguration
public class SampleTestConfiguration {

    @Bean(name = "discardChannel")
    public MessageChannel invalidDiscardChannel() {
        PublishSubscribeChannel publishSubscribeChannel = new PublishSubscribeChannel();
        publishSubscribeChannel.setComponentName("discardChannel");
        return publishSubscribeChannel;
    }
}

public class SampleIntegrationTest{
    @Autowired
    @Qualifier("discardChannel")
    MessageChannel discardChannel;

    @Test
    public void testInvalidMessage() {
        // 触发业务流程

        List<Message<?>> messages = new CopyOnWriteArrayList<>();
        ((PublishSubscribeChannel)discardChannel).subscribe(messages::add);

        await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> {
            // 其他业务断言
            assertTrue(messages.size() > 0);
            Message<?> message = messages.get(0);
            assertNotNull(message);
            // 可添加消息内容、头信息等断言
        });
    }    
}

方案2：通过ChannelInterceptor拦截消息做断言 #

自定义ChannelInterceptor，在消息接收完成后执行断言逻辑，结合CountDownLatch处理异步等待：

private static class PostCompleteAsserter implements ChannelInterceptor {
            
    private final Consumer<Message<?>> assertionLogic;

    public PostCompleteAsserter(Consumer<Message<?>> assertionLogic) {
        this.assertionLogic = assertionLogic;
    }

    @Override
    public void afterReceiveCompletion(@Nullable Message<?> message, MessageChannel channel, @Nullable Exception ex) {
        if (message != null) {
            assertionLogic.accept(message);
        }
    }
}

public class SampleIntegrationTest{
    @Autowired
    @Qualifier("discardChannel")
    MessageChannel discardChannel;

    @Test
    public void testInvalidMessage() {
        // 触发业务流程

        CountDownLatch latch = new CountDownLatch(1);
        ChannelInterceptor asserter = new PostCompleteAsserter(message -> {
            assertNotNull(message);
            // 其他消息内容断言
            latch.countDown();
        });

        ((DirectChannel)discardChannel).addInterceptor(asserter);
        
        assertTrue(latch.await(10, TimeUnit.SECONDS));
    }
}

其他推荐方案与最佳实践 #

方案3：使用Spring Integration Test的TestChannelInterceptor #

Spring Integration Test模块提供了官方的TestChannelInterceptor，内置消息收集功能，无需自定义拦截器：

@SpringBootTest
public class SampleIntegrationTest{
    @Autowired
    @Qualifier("discardChannel")
    DirectChannel discardChannel;

    @Test
    public void testInvalidMessage() {
        TestChannelInterceptor interceptor = new TestChannelInterceptor();
        discardChannel.addInterceptor(interceptor);

        // 触发业务流程

        await().atMost(10, TimeUnit.SECONDS).until(() -> interceptor.getMessages().size() > 0);
        Message<?> discardedMessage = interceptor.getMessages().get(0);
        assertNotNull(discardedMessage);
        // 断言消息内容、头信息等
    }
}

方案4：Mock DiscardChannel的处理端点 #

如果DiscardChannel最终连接的是业务处理Bean（如日志、持久化组件），可以用@SpyBean替换该Bean，直接验证其调用情况：

@SpringBootTest
public class SampleIntegrationTest{
    @SpyBean
    private DiscardMessageHandler discardMessageHandler;

    @Test
    public void testInvalidMessage() {
        // 触发业务流程

        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
            verify(discardMessageHandler, times(1)).handleMessage(any(Message.class));
            // 捕获消息并验证内容
            ArgumentCaptor<Message<?>> messageCaptor = ArgumentCaptor.forClass(Message.class);
            verify(discardMessageHandler).handleMessage(messageCaptor.capture());
            assertNotNull(messageCaptor.getValue());
        });
    }
}

方案5：使用MockIntegration创建测试通道 #

利用Spring Integration的MockIntegration工具类快速创建测试通道，替换原DiscardChannel，支持同步接收消息：

@TestConfiguration
public class SampleTestConfiguration {

    @Bean(name = "discardChannel")
    public MessageChannel testDiscardChannel() {
        return MockIntegration.mockChannel();
    }
}

public class SampleIntegrationTest{
    @Autowired
    @Qualifier("discardChannel")
    MessageChannel discardChannel;

    @Test
    public void testInvalidMessage() {
        // 触发业务流程

        Message<?> discardedMessage = MockIntegration.mockChannel(discardChannel).receive(10, TimeUnit.SECONDS);
        assertNotNull(discardedMessage);
        // 断言消息内容
    }
}

内容的提问来源于stack exchange，提问作者Krittam Kothari