Spring 反应式 ReactorNettyWebSocketClient 不记录任何内容 [英] Spring reactive ReactorNettyWebSocketClient not logging anything

问题描述

我创建了一个简单的演示应用程序来熟悉新的 WebSocketClient.我在以下位置找到了一个教程:https://stackify.com/reactive-spring-5/(反应式 WebSocket 客户端):

I've created a simple demo application to familiarize myself with the new WebSocketClient. I found a tutorial at: https://stackify.com/reactive-spring-5/ (Reactive WebSocket Clients):

@Bean
CommandLineRunner demo() {

   return args -> {

       Flux<String> input = Flux.<String>generate(sink -> sink.next("This is a test"))
               .delayElements(Duration.ofSeconds(1));

       WebSocketClient client = new ReactorNettyWebSocketClient();
       client.execute(new URI("ws://echo.websocket.org"), session ->
               session.send(input.map(session::textMessage))
                       .thenMany(session.receive().map(WebSocketMessage::getPayloadAsText).log())
                       .then())
               .subscribe();
   };
}

我的应用没有崩溃，但我没有看到任何日志记录，我做错了什么吗?我期望从 echo websocket 服务返回相同的值.

My app doesn't crash but I am seeing no logging at all, am I doing something wrong? I am expecting to get the same value returned from the echo websocket service.

如果我用以下内容替换输入变量，它确实有效:

If I replace the input variable with the following it does work:

Mono<String> input = Mono.just("This is a message");

有什么区别，我怎样才能使它与 String 的通量一起工作?

What is the difference and how can I make it work with a flux of String?

推荐答案

这对我有用:

package codependent;

import org.junit.Test;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
import org.springframework.web.reactive.socket.client.WebSocketClient;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.net.URISyntaxException;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;

public class SomeTest {

    @Test
    public void myTest() throws URISyntaxException, InterruptedException {
        CountDownLatch latch = new CountDownLatch(20);

        EmitterProcessor<Object> output = EmitterProcessor.create();

        Flux<String> input = Flux.<String>generate(sink -> sink.next("This is a test"))
                .delayElements(Duration.ofSeconds(1));

        WebSocketClient client = new ReactorNettyWebSocketClient();
        Mono<Void> sessionMono = client.execute(new URI("ws://echo.websocket.org"), session ->
                session.send(input.map(session::textMessage))
                        .thenMany(session.receive()
                                .map(WebSocketMessage::getPayloadAsText).log()
                                .subscribeWith(output).then()).then());


        output.doOnSubscribe(s -> sessionMono.subscribe())
                .subscribe(i -> {
                    System.out.println("Received " + i);
                    latch.countDown();
                });

        latch.await();
    }
}

日志:

18:14:10.205 [reactor-http-nio-4] DEBUG reactor.ipc.netty.http.client.HttpClient - [id: 0x5f974732, L:/192.168.2.193:60438 - R:echo.websocket.org/174.129.224.73:80] READ: 16B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 81 0e 54 68 69 73 20 69 73 20 61 20 74 65 73 74 |..This is a test|
+--------+-------------------------------------------------+----------------+
18:14:10.205 [reactor-http-nio-4] DEBUG io.netty.handler.codec.http.websocketx.WebSocket08FrameDecoder - Decoding WebSocket Frame opCode=1
18:14:10.205 [reactor-http-nio-4] DEBUG io.netty.handler.codec.http.websocketx.WebSocket08FrameDecoder - Decoding WebSocket Frame length=14
18:14:10.205 [reactor-http-nio-4] INFO reactor.Flux.Map.1 - onNext(This is a test)
Received This is a test

这篇关于Spring 反应式 ReactorNettyWebSocketClient 不记录任何内容的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！