Scala Play 2.6 WebSocket问题：错误仅返回发送方而非广播

我来帮你搞定这个需求！你现在的问题是所有消息（包括错误）都被广播给了所有客户端，而你希望只有处理成功的结果才广播，错误只返回给发送请求的那个客户端。咱们先看看原来代码的问题，再一步步修改。

原代码的问题分析 #

你当前用MergeHub接收所有客户端的消息，统一经过processFlow处理后，再通过BroadcastHub广播给所有人。这种方式没法区分消息来自哪个客户端，所以不管是Right还是Left的结果，都会被全量广播，这显然不符合你的需求。

解决方案思路 #

我们需要把每个客户端的连接流单独处理：

• 对每个客户端的输入消息，单独执行process方法
• 处理结果为Right时，把消息发送到全局广播Hub，让所有客户端都能收到
• 处理结果为Left时，直接把错误消息返回给当前客户端，不进入广播流
• 每个客户端的输出流，要合并“自己的错误消息”和“全局广播的成功消息”

修改后的完整代码 #

import akka.stream.scaladsl.{BroadcastHub, Flow, MergeHub, Source, Sink}
import play.api.mvc.WebSocket

// 全局初始化广播Hub：用来分发所有处理成功的消息
val (broadcastSink, broadcastSource) = {
  MergeHub.source[String]
    .toMat(BroadcastHub.sink[String])(Keep.both)
    .run()
}

def ws = WebSocket.accept[String, String] { request =>
  // 处理当前客户端的输入流：区分成功/错误消息
  val clientInputHandler = Flow[String].flatMapConcat { message =>
    process(message) match {
      case Right(successMsg) =>
        // 成功消息发送到全局广播Sink，所有客户端都会收到
        Source.single(successMsg).runWith(broadcastSink)
        // 返回空流，因为广播源会把这条消息同步回来，避免当前客户端收到两次
        Source.empty[String]
      case Left(errorMsg) =>
        // 错误消息仅返回给当前客户端，不进入广播
        Source.single(errorMsg)
    }
  }

  // 合并当前客户端的错误流和全局广播的成功流，作为最终输出
  val clientOutput = clientInputHandler.merge(broadcastSource)

  // 构建WebSocket Flow：输入交给clientInputHandler，输出用合并后的流
  Flow.fromSinkAndSource(clientInputHandler.to(Sink.ignore), clientOutput)
}

def process(message: String): Either[String, String] = {
  if (message == "error") {
    Left(s"ERROR: $message")
  } else {
    Right(s"$message processed")
  }
}

代码解释 #

1. 全局广播Hub：broadcastSink用来接收所有成功的消息，broadcastSource用来给所有客户端分发这些消息，这个是全局唯一的。
2. 客户端输入处理：每个客户端的输入流单独处理，flatMapConcat用来处理Either结果：
   • 成功时，把消息发送到broadcastSink，然后返回空流（因为广播源会把这条消息再发回给当前客户端，避免重复）
   • 错误时，直接返回包含错误消息的源，只会发给当前客户端
3. 输出流合并：把当前客户端的错误消息流和全局广播的成功消息流合并，这样客户端既能收到自己的错误提示，也能收到所有人的成功结果。

这样修改后，就完全满足你的需求了：错误只返回给发送方，成功结果广播给所有客户端。

内容的提问来源于stack exchange，提问作者jo_wil