2

I am not able to figure out a way to attach a WebSocketHandlerAdapter to a reactor netty server.

Requirements: I want to start a reactor netty server and attach http (REST) endpoints and websocket endpoints to the same server. I have gone through the documentation and some sample demo application mentioned in the documentation. They show how to attach a HttpHandlerAdapter to the the HttpServer using newHandler() function. But when it comes to websockets they switch back to using spring boot and annotation examples. I am not able to find how to attach websockets using functional endpoints.

Please point me in the right direction on how to implement this. 1. how do I attach the websocket adapter to the netty server? 2. Should I use HttpServer or TcpServer?

Note: 1. I am not using spring boot. 2. I am not using annotations. 3. Trying to achieve this only using functional webflux end points.

Sample code:

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
  }
4

2 回答 2

4

不幸的是，如果不运行整个 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 提供了非常好的抽象，您可以以相同的功能方式与它进行交互：

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);
                              }
                      );
于 2018-01-04T10:28:29.520 回答
0

我是这样处理的。并使用本机 reactor-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();
                });
    }
于 2019-03-13T10:14:07.700 回答