Akka Streams通过TCP传输

Question

这里是设置：我想通过TCP连接将从发布者转换为字节流的JSON消息流式传输到远程服务器订阅者。
理想情况下，发布者应该是一个Actor，它会接收内部消息，将它们排队，然后在有未完成的需求时将它们流式传输到订阅者服务器。我理解这需要扩展ActorPublisher类以在需要时使用onNext()发送消息。
我的问题是到目前为止，我只能够发送（接收并正确解码）单个消息到服务器并每次打开新连接。我没有设法弄清楚Akka文档并能够使用正确的TCPFlow和ActorPublisher。
以下是发布者的代码：

def send(message: Message): Unit = {
    val system = Akka.system()
    implicit val sys = system

    import system.dispatcher

    implicit val materializer = ActorMaterializer()

    val address =     Play.current.configuration.getString("eventservice.location").getOrElse("localhost")
    val port = Play.current.configuration.getInt("eventservice.port").getOrElse(9000)

    /*** Try with actorPublisher ***/
    //val result = Source.actorPublisher[Message]    (Props[EventActor]).via(Flow[Message].map(Json.toJson(_).toString.map(ByteString(_))))

    /*** Try with actorRef ***/
    /*val source = Source.actorRef[Message](0, OverflowStrategy.fail).map(
  m => {
    Logger.info(s"Sending message: ${m.toString}")
    ByteString(Json.toJson(m).toString)
  }
)
    val ref = Flow[ByteString].via(Tcp().outgoingConnection(address, port)).to(Sink.ignore).runWith(source)*/

    val result = Source(Json.toJson(message).toString.map(ByteString(_))).
  via(Tcp().outgoingConnection(address, port)).
  runFold(ByteString.empty) { (acc, in) ⇒ acc ++ in }//Handle the future
}

最后，演员的代码非常标准：

import akka.actor.Actor
import akka.stream.actor.ActorSubscriberMessage.{OnComplete, OnError}
import akka.stream.actor.{ActorPublisherMessage, ActorPublisher}

import models.events.Message

import play.api.Logger

import scala.collection.mutable

class EventActor extends Actor with ActorPublisher[Message] {
   import ActorPublisherMessage._
   var queue: mutable.Queue[Message] = mutable.Queue.empty

   def receive = {
      case m: Message =>
         Logger.info(s"EventActor - message received and queued: ${m.toString}")
         queue.enqueue(m)
         publish()

      case Request => publish()

      case Cancel =>
          Logger.info("EventActor - cancel message received")
          context.stop(self)

      case OnError(err: Exception) =>
          Logger.info("EventActor - error message received")
          onError(err)
          context.stop(self)

      case OnComplete =>
          Logger.info("EventActor - onComplete message received")
          onComplete()
          context.stop(self)
   }

    def publish() = {
     while (queue.nonEmpty && isActive && totalDemand > 0) {
     Logger.info("EventActor - message published")
     onNext(queue.dequeue())
   }
 }

如有必要，我可以提供订阅者的代码：

def connect(system: ActorSystem, address: String, port: Int): Unit = {
implicit val sys = system
import system.dispatcher
implicit val materializer = ActorMaterializer()

val handler = Sink.foreach[Tcp.IncomingConnection] { conn =>
  Logger.info("Event server connected to: " + conn.remoteAddress)
  // Get the ByteString flow and reconstruct the msg for handling and then output it back
  // that is how handleWith work apparently
  conn.handleWith(
    Flow[ByteString].fold(ByteString.empty)((acc, b) => acc ++ b).
      map(b => handleIncomingMessages(system, b.utf8String)).
      map(ByteString(_))
  )
}

val connections = Tcp().bind(address, port)
val binding = connections.to(handler).run()

binding.onComplete {
  case Success(b) =>
    Logger.info("Event server started, listening on: " + b.localAddress)
  case Failure(e) =>
    Logger.info(s"Event server could not bind to $address:$port: ${e.getMessage}")
    system.terminate()
}
}

感谢您提供的提示。

- Bill'o

---

请按照 Akka 中指定的双向流程，特别使用其 tcp 示例。http://doc.akka.io/docs/akka/current/scala/io-tcp.html - Andrew Scott Evans

Answer 1

我第一条建议是不要编写自己的队列逻辑。Akka已经提供了这个功能。你也不需要编写自己的Actor，Akka Streams也可以提供。

首先，我们可以创建一个Flow，通过Tcp将您的发布者连接到订阅者。在您的发布者代码中，您只需要创建一次ActorSystem并连接到外部服务器即可。

//this code is at top level of your application

implicit val actorSystem = ActorSystem()
implicit val actorMaterializer = ActorMaterializer()
import actorSystem.dispatcher

val host = Play.current.configuration.getString("eventservice.location").getOrElse("localhost")
val port    = Play.current.configuration.getInt("eventservice.port").getOrElse(9000)

val publishFlow = Tcp().outgoingConnection(host, port)

publishFlow 是一个 Flow，它将输入您想要发送到外部订阅者的 ByteString 数据，并输出来自订阅者的 ByteString 数据：

//  data to subscriber ----> publishFlow ----> data returned from subscriber

下一步是发布者Source。而不是编写自己的Actor，您可以使用Source.actorRef将流"materialize"为ActorRef。实质上，这个流将成为我们稍后使用的一个ActorRef：

//these values control the buffer
val bufferSize = 1024
val overflowStrategy = akka.stream.OverflowStrategy.dropHead

val messageSource = Source.actorRef[Message](bufferSize, overflowStrategy)

我们还需要一个Flow将消息转换为字节字符串。

val marshalFlow = 
  Flow[Message].map(message => ByteString(Json.toJson(message).toString))

最后，我们可以连接所有的部分。由于您没有从外部订阅者那里收到任何数据，我们将忽略来自连接的任何数据:

val subscriberRef : ActorRef = messageSource.via(marshalFlow)
                                            .via(publishFlow)
                                            .runWith(Sink.ignore)     

我们现在可以将这个流视为一个Actor：

val message1 : Message = ???

subscriberRef ! message1

val message2 : Message = ???

subscriberRef ! message2

- Ramón J Romero y Vigil