Akka Streams TCP 套接字客户端终止

Question

我有以下流程：

val actorSource = Source.actorRef(10000, OverflowStrategy.dropHead)

val targetSink = Flow[ByteString]
    .map(_.utf8String)
    .via(new JsonStage())
    .map { json =>
      MqttMessages.jsonToObject(json)
    }
    .to(Sink.actorRef(self, "Done"))

  sourceRef = Some(Flow[ByteString]
    .via(conn.flow)
    .to(targetSink)
    .runWith(actorSource))

在Actor（即 Sink.actorRef 之一）内。 conn.flow 是使用 Tcp().bind(address, port) 的传入 TCP 连接。

当前，当从客户端关闭 tcp 连接时，Sink.actorRef Actor 会继续运行。有没有办法注册客户端终止 tcp 连接以关闭Actor？

编辑： 我尝试按照建议处理这两种情况：

case "Done" =>
  context.stop(self)

case akka.actor.Status.Failure =>
  context.stop(self)

但是当我使用套接字客户端进行测试并取消它时，actor 并没有被关闭。因此，如果 TCP 连接终止，“完成”消息和失败似乎都不会被注册。

这是整个代码：

private var connection: Option[Tcp.IncomingConnection] = None
private var mqttpubsub: Option[ActorRef] = None
private var sourceRef: Option[ActorRef] = None

private val sdcTopic = "out"
private val actorSource = Source.actorRef(10000, OverflowStrategy.dropHead)

implicit private val system = context.system
implicit private val mat = ActorMaterializer.create(context.system)

override def receive: Receive = {

case conn: Tcp.IncomingConnection =>
  connection = Some(conn)

  mqttpubsub = Some(context.actorOf(Props(classOf[MqttPubSub], PSConfig(
    brokerUrl = "tcp://127.0.0.1:1883", //all params is optional except brokerUrl
    userName = null,
    password = null,
    //messages received when disconnected will be stash. Messages isOverdue after stashTimeToLive will be discard
    stashTimeToLive = 1.minute,
    stashCapacity = 100000, //stash messages will be drop first haft elems when reach this size
    reconnectDelayMin = 10.millis, //for fine tuning re-connection logic
    reconnectDelayMax = 30.seconds
  ))))

  val targetSink = Flow[ByteString]
    .alsoTo(Sink.foreach(println))
    .map(_.utf8String)
    .via(new JsonStage())
    .map { json =>
      MqttMessages.jsonToObject(json)
    }
    .to(Sink.actorRef(self, "Done"))

  sourceRef = Some(Flow[ByteString]
    .via(conn.flow)
    .to(targetSink)
    .runWith(actorSource))

case msg: MqttMessages.MqttMessage =>
  processMessage(msg)

case msg: Message =>
  val jsonMsg = JsonParser(msg.payload).asJsObject
  val mqttMsg = MqttMessages.jsonToObject(jsonMsg)

  try {
    sourceRef.foreach(_ ! ByteString(msg.payload))
  } catch {
    case e: Throwable => e.printStackTrace()
  }


case SubscribeAck(Subscribe(topic, self, qos), fail) =>

case "Done" =>
  context.stop(self)

case akka.actor.Status.Failure =>
  context.stop(self)
}

Answer 1

Actor 继续运行

你是指哪个演员，你在Sink.actorRef注册的那个？如果是，那么要在流关闭时将其关闭，您需要处理其中的"Done" 和akka.actor.Status.Failure 消息并显式调用context.stop(self)。当流关闭成功时会发送"Done"消息，如果有错误则会发送Status.Failure。

有关更多信息，请参阅Sink.actorRef API 文档，它们解释了终止语义。

Answer 2

我最终创建了另一个 Stage，它只传递元素，但如果上游关闭，则会向下一个流发出附加消息：

class TcpStage extends GraphStage[FlowShape[ByteString, ByteString]] {

   val in = Inlet[ByteString]("TCPStage.in")
   val out = Outlet[ByteString]("TCPStage.out")
   override val shape = FlowShape.of(in, out)

   override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) {

     setHandler(out, new OutHandler {
       override def onPull(): Unit = {
         if (isClosed(in)) emitDone()
         else pull(in)
       }
     })
    setHandler(in, new InHandler {
      override def onPush(): Unit = {
        push(out, grab(in))
      }

      override def onUpstreamFinish(): Unit = {
        emitDone()
        completeStage()
      }
    })

    private def emitDone(): Unit = {
      push(out, ByteString("{ }".getBytes("utf-8")))
    }
  }
}

然后在我的流程中使用：

  val targetSink = Flow[ByteString]
    .via(new TcpStage())
    .map(_.utf8String)
    .via(new JsonStage())
    .map { json =>
      MqttMessages.jsonToObject(json)
    }
    .to(Sink.actorRef(self, MqttDone))

  sourceRef = Some(Flow[ByteString]
    .via(conn.flow)
    .to(targetSink)
    .runWith(actorSource))