Spring Integration测试报错：ReactiveStreamsConsumer的subscriber不能为空

报错信息 #

IllegalArgumentException: 'subscriber' must not be null in the "...endpoint"

测试代码 #

@Test
public void test() throws InterruptedException {
    ArgumentCaptor<Message<?>> captor = messageArgumentCaptor();
    CountDownLatch receiveLatch = new CountDownLatch(1);
    MessageHandler mockMessageHandler = mockMessageHandler(captor).handleNext(m -> receiveLatch.countDown());
    this.mockIntegrationContext
            .substituteMessageHandlerFor(
                    "test2.org.springframework.integration.config.ConsumerEndpointFactoryBean#1",
                    mockMessageHandler);
    this.integrationFlowWithReactiveConsumerHandler.getInputChannel().send(new GenericMessage<>("test2"));
    assertThat(receiveLatch.await(10, TimeUnit.SECONDS)).isTrue();
    verify(mockMessageHandler).handleMessage(any());
    assertThat(captor.getValue().getPayload())
            .isEqualTo("reactive-message-text");
}

报错触发点 #

报错发生在MockIntegrationContext.java的substituteMessageHandlerFor方法中，当处理ReactiveStreamsConsumer类型端点时，代码对subscriber字段做了非空校验：

public void substituteMessageHandlerFor(String consumerEndpointId, // NOSONAR - complexity
        MessageHandler mockMessageHandler, boolean autoStartup) {

    Object endpoint = this.beanFactory.getBean(consumerEndpointId, IntegrationConsumer.class);
    if (autoStartup && endpoint instanceof Lifecycle) {
        ((Lifecycle) endpoint).stop();
    }
    DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(endpoint);
    Object targetMessageHandler = directFieldAccessor.getPropertyValue(HANDLER);
    Assert.notNull(targetMessageHandler, () -> "'handler' must not be null in the: " + endpoint);
    if (endpoint instanceof ReactiveStreamsConsumer) {
        Object targetSubscriber = directFieldAccessor.getPropertyValue("subscriber");
        Assert.notNull(targetSubscriber, () -> "'subscriber' must not be null in the: " + endpoint);

问题根源 #

处理器端点Bean为...:reactive-outbound-channel-adapter，通过ReactiveStreamsConsumer构造函数实例化时，subscriber字段默认被设为null，只有在端点启动（调用start()方法）时才会完成初始化：

/**
 * Instantiate an endpoint based on the provided {@link MessageChannel} and {@link ReactiveMessageHandler}.
 * @param inputChannel the channel to consume in reactive manner.
 * @param reactiveMessageHandler the {@link ReactiveMessageHandler} to process messages.
 * @since 5.3
 */
public ReactiveStreamsConsumer(MessageChannel inputChannel, ReactiveMessageHandler reactiveMessageHandler) {
    Assert.notNull(inputChannel, "'inputChannel' must not be null");
    this.inputChannel = inputChannel;
    this.handler = new ReactiveMessageHandlerAdapter(reactiveMessageHandler);
    this.reactiveMessageHandler = reactiveMessageHandler;
    this.publisher = IntegrationReactiveUtils.messageChannelToFlux(inputChannel);
    this.subscriber = null;
    this.lifecycleDelegate =
            reactiveMessageHandler instanceof Lifecycle ? (Lifecycle) reactiveMessageHandler : null;
}

解决方案 #

方案1：确保端点先启动再替换处理器 #

修改测试代码，先启动目标端点让subscriber完成初始化，再调用替换方法：

@Test
public void test() throws InterruptedException {
    ArgumentCaptor<Message<?>> captor = messageArgumentCaptor();
    CountDownLatch receiveLatch = new CountDownLatch(1);
    MessageHandler mockMessageHandler = mockMessageHandler(captor).handleNext(m -> receiveLatch.countDown());
    
    // 先获取端点并启动
    IntegrationConsumer endpoint = this.beanFactory.getBean("test2.org.springframework.integration.config.ConsumerEndpointFactoryBean#1", IntegrationConsumer.class);
    if (endpoint instanceof Lifecycle) {
        ((Lifecycle) endpoint).start();
    }
    
    this.mockIntegrationContext
            .substituteMessageHandlerFor(
                    "test2.org.springframework.integration.config.ConsumerEndpointFactoryBean#1",
                    mockMessageHandler);
    this.integrationFlowWithReactiveConsumerHandler.getInputChannel().send(new GenericMessage<>("test2"));
    assertThat(receiveLatch.await(10, TimeUnit.SECONDS)).isTrue();
    verify(mockMessageHandler).handleMessage(any());
    assertThat(captor.getValue().getPayload())
            .isEqualTo("reactive-message-text");
}

方案2：直接替换ReactiveMessageHandler而非MessageHandler #

绕过substituteMessageHandlerFor对subscriber的校验，通过DirectFieldAccessor直接替换ReactiveStreamsConsumer中的核心处理器：

@Test
public void test() throws InterruptedException {
    ArgumentCaptor<Message<?>> captor = messageArgumentCaptor();
    CountDownLatch receiveLatch = new CountDownLatch(1);
    
    // 创建自定义的ReactiveMessageHandler模拟实现
    ReactiveMessageHandler mockReactiveHandler = message -> {
        receiveLatch.countDown();
        captor.setValue(message);
        return Mono.empty();
    };
    
    // 获取端点并替换reactiveMessageHandler
    ReactiveStreamsConsumer endpoint = this.beanFactory.getBean("test2.org.springframework.integration.config.ConsumerEndpointFactoryBean#1", ReactiveStreamsConsumer.class);
    DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint);
    accessor.setPropertyValue("reactiveMessageHandler", mockReactiveHandler);
    // 同时替换适配器中的关联处理器
    ReactiveMessageHandlerAdapter adapter = (ReactiveMessageHandlerAdapter) accessor.getPropertyValue("handler");
    new DirectFieldAccessor(adapter).setPropertyValue("handler", mockReactiveHandler);
    
    this.integrationFlowWithReactiveConsumerHandler.getInputChannel().send(new GenericMessage<>("test2"));
    assertThat(receiveLatch.await(10, TimeUnit.SECONDS)).isTrue();
    assertThat(captor.getValue().getPayload())
            .isEqualTo("reactive-message-text");
}

方案3：升级Spring Integration版本 #

如果使用的是较旧版本，该问题可能是已知bug，已在5.5.x及以上版本中修复，升级到最新稳定版即可解决。

内容的提问来源于stack exchange，提问作者Liam Zhang