播放框架 akka 流 websocket 处理消息被发送到死信答案

【问题标题】：Play framework akka stream websocket handling message get sent to deadletters播放框架 akka 流 websocket 处理消息被发送到死信
【发布时间】：2023-03-04 18:26:02
【问题描述】：

我正试图围绕 akka 流和处理 Web 套接字的方式展开思考，但有些事情对我来说很清楚。对于初学者，我试图完成从某个客户端到服务器的单向通信以及同一服务器和其他某个客户端之间的通信。

client1 -----> Server <------> client2

我正在查看here 提供的示例。生成的代码如下所示：

1) 从控制器开始

class Test @Inject()(@Named("connManager") myConnectionsManager: ActorRef, cc: ControllerComponents)
                (implicit val actorSystem: ActorSystem,
                 val mat: Materializer,
                 implicit val executionContext: ExecutionContext)
extends AbstractController(cc) {


private def wsFutureFlow(id: String): Future[Flow[String, String, NotUsed]] = {
    implicit val timeout: Timeout = Timeout(5.seconds)
    val future = myConnectionsManager ? CreateRemote(id)
    val futureFlow = future.mapTo[Flow[String, String, NotUsed]]
    futureFlow
}

private def wsFutureLocalFlow: Future[Flow[String, String, NotUsed]] =   {
    implicit val timeout: Timeout = Timeout(5.seconds)
    val future = myConnectionsManager ? CreateLocal
    val futureFlow = future.mapTo[Flow[String, String, NotUsed]]
    futureFlow
}

def ws: WebSocket = WebSocket.acceptOrResult[String, String] {

    rh =>
        wsFutureFlow(rh.id.toString).map { flow =>
            Right(flow)
        }
}

def wsLocal: WebSocket = WebSocket.acceptOrResult[String, String] {

    _ =>
        wsFutureLocalFlow.map { flow =>
            Right(flow)
        }
}
}

至于连接管理器参与者。这相当于示例中的 UserParentActor。

class MyConnectionsManager @Inject()(childFactory: MyTestActor.Factory)
                                (implicit ec: ExecutionContext, mat: Materializer) extends Actor with InjectedActorSupport {

import akka.pattern.{ask, pipe}

implicit val timeout: Timeout = Timeout(2.seconds)

override def receive: Receive = {
    case CreateRemote(x) =>
        val child = injectedChild(childFactory(), s"remote-$x")
        context.watch(child)
        privatePipe(child)
    case CreateLocal =>
        val child = injectedChild(childFactory(), "localConnection")
        context.become(onLocalConnected(child))
        privatePipe(child)
    case Terminated(child) =>
        println(s"${child.path.name} terminated...")
}

def onLocalConnected(local: ActorRef): Receive = {
    case CreateRemote(x) =>
        val child = injectedChild(childFactory(), s"remote-$x")
        context.watch(child)
        privatePipe(child)
    case x: SendToLocal => local ! x
}

private def privatePipe(child: ActorRef) = {
    val future = (child ? Init).mapTo[Flow[String, String, _]]
    pipe(future) to sender()
    () // compiler throws exception without this: non-unit value discarded
    }
}

MyTestActor 看起来像这样：

class MyTestActor @Inject()(implicit mat: Materializer, ec: ExecutionContext) extends Actor {

val source: Source[String, Sink[String, NotUsed]] = MergeHub.source[String]
  .recoverWithRetries(-1, { case _: Exception => Source.empty })

private val jsonSink: Sink[String, Future[Done]] = Sink.foreach { json =>
    println(s"${self.path.name} got message:  $json")

    context.parent ! SendToLocal(json)
}

private lazy val websocketFlow: Flow[String, String, NotUsed] = {
    Flow.fromSinkAndSourceCoupled(jsonSink, source).watchTermination() { (_, termination) =>
        val name = self.path.name
        termination.foreach(_ => context.stop(self))
        NotUsed
    }
}

def receive: Receive = {

    case Init =>
        println(s"${self.path.name}: INIT")
        sender ! websocketFlow
    case SendToLocal(x) =>
        println(s"Local got from remote: $x")
    case msg: String => sender ! s"Actor got message: $msg"
    }
}

除了接收器和源如何实际连接到演员之外，我不明白的是以下内容。当我启动我的系统时，我会向演员发送一些消息。但是，在我关闭与名为 remote 的参与者的连接并继续向名为“localConnection”的参与者发送消息后，这些消息将发送到 DeadLetters：

[info] Done compiling.
[info] 15:49:20.606 - play.api.Play - Application started (Dev)
localConnection: INIT
localConnection got message:  test data
Local got from remote: test data
localConnection got message:  hello world
Local got from remote: hello world
remote-133: INIT
remote-133 got message:  hello world
Local got from remote: hello world
remote-133 got message:  hello from remote
Local got from remote: hello from remote
[error] 15:50:24.449 - a.a.OneForOneStrategy - Monitored actor [Actor[akka://application/user/connManager/remote-133#-998945083]] terminated
akka.actor.DeathPactException: Monitored actor [Actor[akka://application/user/connManager/remote-133#-998945083]] terminated
deadLetters got message:  hello local

我认为这是因为引发了异常...谁能向我解释为什么将消息发送到 DeadLetters？除此之外，我想知道为什么在privatePipe 末尾没有返回“()”的情况下，我一直得到编译器异常？

另外，我应该做些不同的事情吗？

【问题讨论】：

标签： scala playframework akka

【解决方案1】：

我意识到抛出异常是因为我忘记在 MyConnectionsManager 角色的新行为中处理 Terminated 消息。

def onLocalConnected(local: ActorRef): Receive = {
    case CreateRemote(x) =>
        val child = injectedChild(childFactory(), s"remote-$x")
        context.watch(child)
        privatePipe(child)
    case Terminated(child) => println(s"${child.path.name} terminated...")
    case x: SendToLocal => local ! x
}

它现在似乎正在工作。

【讨论】：