Sending output of one websocket client as input to another

user_7457872 0
Viewed 211

Quick search on SO failed to find me a similar question so here we go

I basically want RSocket's requestChannel syntax with Webflux so I am able to process the received Flux outside of WebSocketClient.execute() method and write something like this (with session being opened only when the returned flux is subscribed to, proper error propagation, automatic completion and closing of the WS session when both inbound and outbound fluxes are complete - either completed by the server side or cancelled by the consumer)

service /f wraps its received string messages in 'f(...)': 'str' -> 'f(str)'

service /g does the same with 'g(...)' and the following test passes:

    private final DataBufferFactory dataBufferFactory = new DefaultDataBufferFactory();

    private WebSocketMessage serializeString(final String text) {
        return new WebSocketMessage(Type.TEXT, dataBufferFactory.wrap(text.getBytes(StandardCharsets.UTF_8)));
    }

    @Test
    void test() {
        var requests = 5;
        var input = Flux.range(0, requests).map(String::valueOf);

        var wsClient = new ReactorNettyWebSocketClient(
            HttpClient.from(TcpClient.create(ConnectionProvider.newConnection())));

        var f = requestChannel(wsClient, fUri, input.map(this::serializeString))
            .map(WebSocketMessage::getPayloadAsText);

        var g = requestChannel(wsClient, gUri, f.map(this::serializeString))
            .map(WebSocketMessage::getPayloadAsText);

        var responses = g.take(requests);

        var expectedResponses = Stream.range(0, requests)
                                           .map(i -> "g(f(" + i + "))")
                                           .toJavaArray(String[]::new);

        StepVerifier.create(responses)
                    .expectSubscription()
                    .expectNext(expectedResponses)
                    .verifyComplete();
    }
project-reactor spring-webflux java
1 Answers

And this seems to work for me... so far

    public static Flux<WebSocketMessage> requestChannel(
        WebSocketClient wsClient, URI uri, Flux<WebSocketMessage> outbound) {

        CompletableFuture<Flux<WebSocketMessage>> recvFuture = new CompletableFuture<>();
        CompletableFuture<Integer> consumerDoneCallback = new CompletableFuture<>();

        var executeMono = wsClient.execute(uri,
            wss -> {
                recvFuture.complete(wss.receive().log("requestChannel.receive " + uri, Level.FINE));
                return wss.send(outbound)
                          .and(Mono.fromFuture(consumerDoneCallback));
            }).log("requestChannel.execute " + uri, Level.FINE);

        return Mono.fromFuture(recvFuture)
                   .flatMapMany(recv -> recv.doOnComplete(() -> consumerDoneCallback.complete(1)))
                   .mergeWith(executeMono.cast(WebSocketMessage.class));
    }

Rather interested if there're any flaws with this solution I haven't stumbled upon yet

user_74578720
Related