您将如何更改此 Akka Streams 示例以获取物化值 Future[ServerBinding]？答案

【问题标题】：How would you change this Akka Streams example to get the materialized value Future[ServerBinding]?您将如何更改此 Akka Streams 示例以获取物化值 Future[ServerBinding]？
【发布时间】：2016-03-03 04:22:16
【问题描述】：

http://doc.akka.io/docs/akka/2.4.2/scala/stream/stream-io.html 的第二个示例如下。我添加了host 和port 的定义。

  val host = "localhost"
  val port = 4444
  val connections: Source[IncomingConnection, Future[ServerBinding]] =
    Tcp().bind(host, port)

  connections runForeach { (connection: IncomingConnection) =>
    println(s"New connection from: ${connection.remoteAddress}")

    val echo = Flow[ByteString]
      .via(Framing.delimiter(
        ByteString("\n"),
        maximumFrameLength = 256,
        allowTruncation = true))
      .map(_.utf8String)
      .map(_ + "!!!\n")
      .map(ByteString(_))

    connection.handleWith(echo)
  }

您将如何更改示例以便也获得Future[ServerBinding]，以便您可以记录成功的绑定并处理绑定错误，例如，如果端口已经在用吗？

【问题讨论】：

标签： scala akka akka-stream

【解决方案1】：

您应该能够摆脱runForeach 的便利性，它隐藏了所涉及的Sink 的详细信息，而是使用显式Sink。像这样的：

  val host = "localhost"
  val port = 4444
  val connections: Source[IncomingConnection, Future[ServerBinding]] =
    Tcp().bind(host, port)


  val sink = Sink.foreach[IncomingConnection]{ connection =>
    println(s"New connection from: ${connection.remoteAddress}")

    val echo = Flow[ByteString]
      .via(Framing.delimiter(
        ByteString("\n"),
        maximumFrameLength = 256,
        allowTruncation = true))
      .map(_.utf8String)
      .map(_ + "!!!\n")
      .map(ByteString(_))

    connection.handleWith(echo)    
  }

  val serverBinding:Future[ServerBinding] = connections.to(sink).run

【讨论】：