spring webflux:将websocket适配器连接到reactor-netty服务器的纯功能方式

Question

我不能想出一种方法将WebSocketHandlerAdapter连接到反应器网络服务器上。

要求:我想启动一个reactor netty服务器，并将http (REST)端点和websocket端点连接到同一服务器。我已经阅读了文档和文档中提到的一些示例演示应用程序。它们展示了如何使用newHandler()函数将HttpHandlerAdapter附加到HttpServer。但当涉及到websockets时，他们又切换回使用spring boot和注解示例。我无法找到如何使用功能端点附加websockets。

关于如何实现这一点，请给我指出正确的方向。1.如何将websocket适配器连接到netty服务器? 2.我应该使用HttpServer还是TcpServer？

注意: 1.我没有使用spring boot。2.我没有使用注释。3.尝试仅使用功能webflux端点来实现这一点。

示例代码：

public HandlerMapping webSocketMapping() 
{
  Map<String, WebSocketHandler> map = new HashMap<>();
  map.put("/echo", new EchoTestingWebSocketHandler());
  SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
  mapping.setUrlMap(map);
  mapping.setOrder(-1);
  return mapping;
}
public WebSocketHandlerAdapter wsAdapter() 
{
  HandshakeWebSocketService wsService = new HandshakeWebSocketService(new ReactorNettyRequestUpgradeStrategy());
  return new WebSocketHandlerAdapter(wsService);
}

  protected void startServer(String host, int port) 
  {
    HttpServer server = HttpServer.create(host, port);
    server.newHandler(wsAdapter()).block();    //how do I attach the websocket adapter to the netty server
  }

Answer 1

不幸的是，在不运行整个SpringBootApplication的情况下，没有简单的方法可以做到这一点。否则，您将需要自己编写整个Spring WebFlux处理程序层次结构。考虑使用SpringBootApplication组成功能路由：

    @SpringBootApplication
    public class WebSocketApplication {

        public static void main(String[] args) {
            SpringApplication.run(WebSocketApplication.class, args);
        }


        @Bean
        public RouterFunction<ServerResponse> routing() {
            return route(
                    POST("/api/orders"),
                    r -> ok().build()
            );
        }

        @Bean
        public HandlerMapping wsHandlerMapping() {
            HashMap<String, WebSocketHandler> map = new HashMap<>();

            map.put("/ws", new WebSocketHandler() {
                @Override
                public Mono<Void> handle(WebSocketSession session) {
                    return session.send(
                            session.receive()
                                  .map(WebSocketMessage::getPayloadAsText)
                                  .map(tMessage -> "Response From Server: " + tMessage)
                                  .map(session::textMessage)
                    );
                }
            });

            SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
            mapping.setUrlMap(map);
            mapping.setOrder(-1);
            return mapping;
        }

        @Bean
        HandlerAdapter wsHandlerAdapter() {
            return new WebSocketHandlerAdapter();
        }
    }

如果SpringBoot下拉不是这样，则为大小写

尝试考虑与ReactorNetty的直接交互。Reactor Netty围绕原生Netty提供了pritty良好的抽象，您可以以相同的函数方式与其交互：

ReactorHttpHandlerAdapter handler =
                    new ReactorHttpHandlerAdapter(yourHttpHandlers);

            HttpServer.create()
                      .startRouterAndAwait(routes -> {
                                  routes.ws("/pathToWs", (in, out) -> out.send(in.receive()))
                                        .file("/static/**", ...)
                                        .get("**", handler)
                                        .post("**", handler)
                                        .put("**", handler)
                                        .delete("**", handler);
                              }
                      );

Answer 2

我是这样处理的。并使用本地反应器-netty

routes.get(rootPath, (req, resp)->{
        // doFilter check the error
        return this.doFilter(request, response, new RequestAttribute())
                .flatMap(requestAttribute -> {
                    WebSocketServerHandle handleObject = injector.getInstance(GameWsHandle.class);
                    return response
                        .header("content-type", "text/plain")
                        .sendWebsocket((in, out) ->
                            this.websocketPublisher3(in, out, handleObject, requestAttribute)
                        );
                });
    })

private Publisher<Void> websocketPublisher3(WebsocketInbound in, WebsocketOutbound out, WebSocketServerHandle handleObject, RequestAttribute requestAttribute) {
        return out
                .withConnection(conn -> {
                    // on connect
                    handleObject.onConnect(conn.channel());
                    conn.channel().attr(AttributeKey.valueOf("request-attribute")).set(requestAttribute);
                    conn.onDispose().subscribe(null, null, () -> {
                            conn.channel().close();
                            handleObject.disconnect(conn.channel());
                            // System.out.println("context.onClose() completed");
                        }
                    );
                    // get message
                    in.aggregateFrames()
                            .receiveFrames()
                            .map(frame -> {
                                if (frame instanceof TextWebSocketFrame) {
                                    handleObject.onTextMessage((TextWebSocketFrame) frame, conn.channel());
                                } else if (frame instanceof BinaryWebSocketFrame) {
                                    handleObject.onBinaryMessage((BinaryWebSocketFrame) frame, conn.channel());
                                } else if (frame instanceof PingWebSocketFrame) {
                                    handleObject.onPingMessage((PingWebSocketFrame) frame, conn.channel());
                                } else if (frame instanceof PongWebSocketFrame) {
                                    handleObject.onPongMessage((PongWebSocketFrame) frame, conn.channel());
                                } else if (frame instanceof CloseWebSocketFrame) {
                                    conn.channel().close();
                                    handleObject.disconnect(conn.channel());
                                }
                                return "";
                            })
                            .blockLast();
                });
    }