如何关闭Akka流？ [英] How does one close an Akka stream?
本文介绍了如何关闭Akka流？的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！
问题描述
我正在将一些实验喷雾代码更新为akka-stream（2.0.2），因此我将逐步阅读他的文档。我的需求之一是，如果我在流中检测到协议冲突，则需要立即关闭流并启动客户端。
I am updating some experimental spray code to akka-stream (2.0.2), so I'm going through he documentation little by little. One of the needs I have is that if I detect a protocol violation in my stream, I need to close the stream immediately and kick off the client.
什么是适当的立即从Flow内部关闭（终止）流的方法？
What is the proper way of immediately closing (terminating) the stream from inside a Flow?
推荐答案
使用 PushStage ：
import akka.stream.stage._
val closeStage = new PushStage[Tpe, Tpe] {
override def onPush(elem: Tpe, ctx: Context[Tpe]) = elem match {
case elem if shouldCloseStream ⇒
// println("stream closed")
ctx.finish()
case elem ⇒
ctx.push(elem)
}
}
您可以将 PushStage 与 Flow 通过 transform 方法：
Flow[Tpe]
.map(...)
.transform(() ⇒ closeStage)
.map(...)
这篇关于如何关闭Akka流？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
查看全文
