如何关闭阿卡流？

Question

我正在更新一些实验性喷雾代码，以使阿卡流（2.0.2），所以我一点一点地通过他的文档。我的一个需求是，如果我在流中检测到协议违规，我需要立即关闭流并启动客户端。

什么是立即关闭（终止）流内流的正确方法？

Answer 1

使用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) 
    } 
} 

你可以通过transform方法与Flow一个PushStage结合：

Flow[Tpe] 
.map(...) 
.transform(() ⇒ closeStage) 
.map(...)