如何异步接受 WebSocket？

Posted 2023-02-16

【中文标题】如何异步接受 WebSocket？

【英文标题】：How to accept a WebSocket asynchronously?

【发布时间】：2019-08-21 14:23:59

【问题描述】：

我有一个处理 WebSocket 请求的 Play 应用程序。路由文件包含这一行：

GET  /testsocket  controllers.HomeController.defaultRoomSocket

一个已经工作的同步版本如下所示：（改编自 2.7.x 文档）

public WebSocket defaultRoomSocket() 
    return WebSocket.Text.accept(
        request -> ActorFlow.actorRef(MyWebSocketActor::props, actorSystem, materializer));
  

如https://www.playframework.com/documentation/2.7.x/JavaWebSockets#Accepting-a-WebSocket-asynchronously 中所述，我将签名更改为

public CompletionStage<WebSocket> defaultRoomSocket()
   //returning a CompletionStage here, using the "ask pattern"
   //to get the needed Flow from an other Actor

从这里我遇到了以下问题：

Cannot use a method returning java.util.concurrent.CompletionStage[play.mvc.WebSocket] as a Handler for requests

此外，正如文档所暗示的，“WebSocket”没有 TypeParameter。接受 WebSocket 异步请求的适当方式是什么？

【参考方案1】：

文档确实需要更新，我认为#5055的websockets重构中遗漏了一些内容。

要获得异步处理，您应该使用acceptOrResult 方法，该方法将CompletionStage 作为返回类型而不是流。然后可以使用函数式编程助手 (F.Either) 返回 Result 或 Akka Flow。其实accept方法是这样实现的：

public WebSocket accept(Function<Http.RequestHeader, Flow<In, Out, ?>> f) 
   return acceptOrResult(
       request -> CompletableFuture.completedFuture(F.Either.Right(f.apply(request))));

如您所见，它所做的只是使用 completedFuture 调用异步版本。

要完全使其异步并达到我认为您想要实现的目标，您可以执行以下操作：

public WebSocket ws() 
    return WebSocket.Json.acceptOrResult(request -> 
        if (sameOriginCheck(request)) 
            final CompletionStage<Flow<JsonNode, JsonNode, NotUsed>> future = wsFutureFlow(request);
            final CompletionStage<Either<Result, Flow<JsonNode, JsonNode, ?>>> stage = future.thenApply(Either::Right);
            return stage.exceptionally(this::logException);
         else 
            return forbiddenResult();
        
    );


@SuppressWarnings("unchecked")
private CompletionStage<Flow<JsonNode, JsonNode, NotUsed>> wsFutureFlow(Http.RequestHeader request) 
    long id = request.asScala().id();
    UserParentActor.Create create = new UserParentActor.Create(Long.toString(id));

    return ask(userParentActor, create, t).thenApply((Object flow) -> 
        final Flow<JsonNode, JsonNode, NotUsed> f = (Flow<JsonNode, JsonNode, NotUsed>) flow;
        return f.named("websocket");
    );


private CompletionStage<Either<Result, Flow<JsonNode, JsonNode, ?>>> forbiddenResult() 
    final Result forbidden = Results.forbidden("forbidden");
    final Either<Result, Flow<JsonNode, JsonNode, ?>> left = Either.Left(forbidden);

    return CompletableFuture.completedFuture(left);


private Either<Result, Flow<JsonNode, JsonNode, ?>> logException(Throwable throwable) 
    logger.error("Cannot create websocket", throwable);
    Result result = Results.internalServerError("error");
    return Either.Left(result);

（这取自play-java-websocket-example，可能很有趣）

如您所见，在返回 websocket 连接或 HTTP 状态之前，它首先经过几个阶段。

【讨论】：

感谢您为示例提供提示。您的猜测是正确的，这完全解决了我想要实现的目标。至少对我来说，这个例子更好地展示了如何使用acceptOrResult（与文档/教程相比）