如何停止反应堆 websocket 连接答案

【问题标题】：How to stop reactor websocket connection如何停止反应堆 websocket 连接
【发布时间】：2021-10-29 19:27:34
【问题描述】：

鉴于 reactor 文档中的这个示例：

import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;
import reactor.core.publisher.Flux;
import reactor.netty.http.client.HttpClient;

public class Application {

    public static void main(String[] args) {
        HttpClient client = HttpClient.create();

        client.websocket()
              .uri("wss://echo.websocket.org")
              .handle((inbound, outbound) -> {
                  inbound.receive()
                         .asString()
                         .take(1)
                         .subscribe(System.out::println);

                  final byte[] msgBytes = "hello".getBytes(CharsetUtil.ISO_8859_1);
                  return outbound.send(Flux.just(Unpooled.wrappedBuffer(msgBytes), Unpooled.wrappedBuffer(msgBytes)))
                                 .neverComplete();
              })
              .blockLast();
    }
}

当take(1) 或任何其他条件为真时，如何完全停止和断开与 websocket 的连接？现在它无限期挂起并且不退出

【问题讨论】：

标签： java websocket java-websocket reactor-netty

【解决方案1】：

似乎可以通过删除neverComplete 来修复此示例，如下所示：

import io.netty.util.CharsetUtil;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;

public class Application {

    public static void main(String[] args) {
        HttpClient client = HttpClient.create();

        client.websocket()
        .uri("wss://echo.websocket.org")
        .handle((inbound, outbound) -> {

            return outbound.sendString(Mono.just("hello"), CharsetUtil.UTF_8)
                    .then()
                    .thenMany(inbound.receive()
                            .asString()
                            .take(1)
                            .doOnNext(System.out::println));
        })
        .blockLast();
    }
}

【讨论】：