akkaストリームのライフサイクルを監視する

Question

私はakka-streamのライフサイクルを監視したいと思っています。monitorは私の必要とするものですが、私の監視機能は非同期で、Futureを返すので、モニターが必要ですasyncも同様です。

monitorは、次のシグネチャがあります。

def monitor[Mat2]()(combine: (Mat, FlowMonitor[Out]) ⇒ Mat2): ReprMat[Out, Mat2] 

をしかし、私はのような何か必要があります：

def monitorAsync[Mat2]()(combine: (Mat, FlowMonitor[Out]) ⇒ Future[Mat2]): ReprMat[Out, Mat2] 

をmapAsyncのように、この使用アッカ・ストリームのプリミティブを実装する方法はあります。

私はmapAsync + watchTerminationを使用することができますが、monitorがほぼ必要なときには複雑な解決策のようです。

Answer 1

monitorは、ストリームのマテリアライゼーション後にはFlowMonitorにしかアクセスできないため、私が望むものではありませんでした。

mapAsyncとrecoverでこれを実装しました。私はここに簡素化していますが、このような何か：

val monitor = new Monitor { 
    def onNext: Future[Unit] = ??? 
    def onFailure(cause: Throwable): Future[Unit] = ??? 
    def onFinish: Future[Unit] = ??? 
} 

source.mapAsync { v => 
    monitor.onNext.map(_ => v) 
}.watchTermination() { (mat, doneF) => 
    doneF.flatMap(_ => monitor.onFinish).recoverWith(case ex => monitor.onFailure(ex)) 
}