¿Cuál es la forma correcta de consumir el flujo entrante de ReactorNettyWebSocketClient?

Intento aprender la programación reactiva con Spring Reactor Webflux. Particularmente en el siguiente ejemplo, trato de entender cómo componer correctamente el flujo de procesamiento para consumir el flujo entrante de una conexión WebSocket utilizando ReactorNettyWebSocketClient.

El siguiente fragmento de código funciona correctamente:

WebSocketClient client = new ReactorNettyWebSocketClient();
client.execute(
URI.create(“wss://stream.binance.com:9443/ws/btcusdt@trade”),
session -> {
Flux tradesFlux = session.receive()
.map(WebSocketMessage::getPayloadAsText)
.doOnNext(event -> log.info(event));

            return tradesFlux.then();
        }

).subscribe();

En el fragmento de código anterior, uso doOnNext() en el flujo entrante Flux – tradesFlux para consumir cada evento que llega. Funciona bien con el fragmento de código anterior, pero según tengo entendido, doOnNext() es un operador de efecto secundario, así que intenté hacer lo siguiente:

WebSocketClient client = new ReactorNettyWebSocketClient();
client.execute(
URI.create(“wss://stream.binance.com:9443/ws/btcusdt@trade”),
session -> {
Flux tradesFlux = session.receive()
.map(WebSocketMessage::getPayloadAsText);

            tradesFlux.subscribe(new Subscriber<string>() {
                @Override
                public void onSubscribe(Subscription s) {
                }

                @Override
                public void onNext(String s) {
                    log.info("replicar operación de trading de Binance {}", s);
                }

                @Override
                public void onError(Throwable t) {
                }

                @Override
                public void onComplete() {
                }
            });

            return tradesFlux.then();
        }

).subscribe();

En el segundo fragmento de código, intenté consumir el flujo entrante con un Subscriber llamando a subscribe en el flujo entrante – tradesFlux. Pero con este fragmento de código, obtuve la siguiente excepción:

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException: Solo se permite un único suscriptor de recepción de conexión.
Causado por: java.lang.IllegalStateException: Solo se permite un único suscriptor de recepción de conexión.

De alguna manera no se puede consumir el flujo entrante Flux usando su propio Subscriber, así que la pregunta que quiero hacer es: ¿es el primer fragmento de código la forma adecuada de consumir el flujo entrante de ReactorNettyWebSocketClient utilizando el operador de efecto secundario doOnNext(), o estoy omitiendo algo aquí?

¡Muchas gracias!
Tung