无法向Mock ReactorNettyWebSocketClient发送Flux消息的测试问题

核心问题分析 #

测试阻塞的本质是Mock的ReactorNettyWebSocketClient.execute()未返回可正常完成的响应式流，同时直接MockWebSocketSession时，没有模拟完整的消息收发响应式逻辑，导致测试一直等待信号触发。

正确的测试方案 #

1. 模拟完整的WebSocket会话交互 #

避免仅Mock客户端外层对象，转而聚焦于模拟WebSocketSession的消息流行为，确保响应式流能正常触发完成信号。

2. 示例代码实现 #

参考生产代码

假设你的业务处理类如下：

@Component
public class WebSocketClientService {
    private final ReactorNettyWebSocketClient webSocketClient;

    public WebSocketClientService(ReactorNettyWebSocketClient webSocketClient) {
        this.webSocketClient = webSocketClient;
    }

    public Flux<String> connectAndConsume(URI uri) {
        return webSocketClient.execute(uri, session ->
                session.receive()
                        .map(WebSocketMessage::getPayloadAsText)
                        .doOnNext(this::processMessage)
                        .thenMany(session.send(Mono.just(session.textMessage("client-ready"))))
                        .then()
                        .flux()
        );
    }

    private void processMessage(String msg) {
        // 业务处理逻辑
    }
}

单元测试代码

import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.adapter.TextMessage;
import reactor.netty.http.client.ReactorNettyWebSocketClient;

import java.net.URI;

public class WebSocketClientServiceTest {

    @Test
    void testWebSocketMessageConsumption() {
        // 1. Mock WebSocketSession
        WebSocketSession mockSession = Mockito.mock(WebSocketSession.class);

        // 2. 模拟服务端发送的消息流（用真实TextMessage减少Mock复杂度）
        WebSocketMessage mockMsg1 = new TextMessage("server-msg-1");
        WebSocketMessage mockMsg2 = new TextMessage("server-msg-2");
        Flux<WebSocketMessage> mockServerMessages = Flux.just(mockMsg1, mockMsg2);
        Mockito.when(mockSession.receive()).thenReturn(mockServerMessages);

        // 3. 模拟客户端发送消息的逻辑，返回完成信号
        Mockito.when(mockSession.send(Mockito.any(Flux.class))).thenReturn(Mono.empty());

        // 4. Mock WebSocketClient，让execute执行传入的会话处理器
        ReactorNettyWebSocketClient mockClient = Mockito.mock(ReactorNettyWebSocketClient.class);
        Mockito.when(mockClient.execute(Mockito.any(URI.class), Mockito.any()))
                .thenAnswer(invocation -> {
                    WebSocketSession.Handler handler = invocation.getArgument(1);
                    return handler.handle(mockSession);
                });

        // 5. 初始化业务类并验证流行为
        WebSocketClientService service = new WebSocketClientService(mockClient);
        StepVerifier.create(service.connectAndConsume(URI.create("ws://test-endpoint")))
                .expectNext("server-msg-1", "server-msg-2")
                .verifyComplete();
    }
}

3. 关键注意事项 #

• 确保Mock流可完成：所有Mono/Flux必须返回带有完成信号的流（如Mono.empty()、Flux.just(...)），避免测试无限等待。
• 优先使用真实消息对象：用TextMessage替代MockWebSocketMessage，减少Mock细节，提升测试真实性。
• 用StepVerifier控制时序：它能明确验证流的元素数量、完成状态，超时后自动终止测试，避免阻塞。

4. 进阶：嵌入式服务器集成测试（可选） #

如果需要更贴近生产的测试场景，可使用WebSocketTestClient配合嵌入式服务器，无需Mock客户端：

import org.junit.jupiter.api.Test;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.client.WebSocketTestClient;

public class WebSocketIntegrationTest {

    @Test
    void testRealWebSocketInteraction() {
        // 模拟服务端处理器
        WebSocketHandler serverHandler = session ->
                session.receive()
                        .map(WebSocketMessage::getPayloadAsText)
                        .thenMany(session.send(Flux.just(
                                session.textMessage("resp-1"),
                                session.textMessage("resp-2")
                        )))
                        .then();

        WebSocketTestClient client = WebSocketTestClient.create(serverHandler);
        client.execute("/ws-path")
                .send(Mono.just(client.textMessage("client-init")))
                .expectTextFrame("resp-1")
                .expectTextFrame("resp-2")
                .verifyComplete();
    }
}

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