提示:这个答案基于akka-stream-experimental
版本2.0-M2
。其他版本的 API 可能略有不同。
关闭连接的一个简单方法是使用PushStage
:
import akka.stream.stage._
val closeClient = new PushStage[String, String] {
override def onPush(elem: String, ctx: Context[String]) = elem match {
case "goodbye" ⇒
// println("Connection closed")
ctx.finish()
case msg ⇒
ctx.push(msg)
}
}
在客户端或服务器端接收到的每个元素(通常是通过Flow
)经历这样一个Stage
成分。在 Akka 中,完整的抽象称为GraphStage
,更多信息可以在官方文档.
With a PushStage
我们可以观察具体传入元素的值,然后相应地转换上下文。在上面的例子中,一旦goodbye
收到消息后,我们完成上下文,否则我们只需通过push
method.
现在,我们可以连接closeClient
通过的任意流分量transform
method:
val connection = Tcp().outgoingConnection(address, port)
val flow = Flow[ByteString]
.via(Framing.delimiter(
ByteString("\n"),
maximumFrameLength = 256,
allowTruncation = true))
.map(_.utf8String)
.transform(() ⇒ closeClient)
.map(_ ⇒ StdIn.readLine("> "))
.map(_ + "\n")
.map(ByteString(_))
connection.join(flow).run()
上面的流程接收到一个ByteString
并返回一个ByteString
,这意味着它可以连接到connection
通过join
方法。在流程内部,我们首先将字节转换为字符串,然后再将它们发送到closeClient
。如果PushStage
没有完成流,该元素在流中转发,在那里它被丢弃并被来自标准输入的一些输入替换,然后通过线路发送回。如果流完成,则阶段组件之后的所有进一步流处理步骤都将被删除 - 流现在已关闭。