ステートフルフローマテリアライゼーション最終状態

Question

処理されたデータの統計を収集し、結果の統計を具体化するakkaストリームフィルタフローを実装しようとしています。

class SFilter[A](p: A => Boolean) extends GraphStage[FlowShape[A, A]] { 
    val in = Inlet[A]("SFilter.in") 
    val out = Outlet[A]("SFilter.out") 
    val shape = FlowShape.of(in, out) 
    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = 
    new GraphStageLogic(shape) { 
     var positive: Long = 0 
     var negative: Long = 0 
     setHandler(in, new InHandler { 
     override def onPush(): Unit = { 
      val elem = grab(in) 
      if (p(elem)) { 
      push(out, elem) 
      positive += 1 
      } else { 
      pull(in) 
      negative += 1 
      } 
     } 
     }) 
     setHandler(out, new OutHandler { 
     override def onPull(): Unit = { 
      pull(in) 
     } 
     }) 
    } 
} 

これまでのところは良いが、私は私のSFilter[A]タイプFlow[A,A,(Long,Long)]であるようになります。どのようにしてコンフルエンスの最後に(positive,negative)を具体化できますか？

Answer 1

Tuple2[Long, Long]は実行中のストリーム自体に依存するため、Tuple2[Long, Long]をマテリアライズすることはできません。ただし、Future[Tuple2[Long, Long]]を実体化することができます。このストリームは、ストリームの完了時に完了します。

編集：通常のフィルターとSFilterを区別できるように、カスタムステージに異なる名前を付けたいとします。ヴィクトルクランの提案に

Answer 2

おかげで、私は以下のソリューションを実装することができました：

class SFilter[A](p: A => Boolean) extends GraphStageWithMaterializedValue[FlowShape[A,A],Future[(Long,Long)]] { 
    val in = Inlet[A]("SFilter.in") 
    val out = Outlet[A]("SFilter.out") 
    val shape = FlowShape.of(in, out) 
    override def createLogicAndMaterializedValue(inheritedAttributes: Attributes) = { 
    val result = Promise[(Long,Long)]() 
    val logic = new GraphStageLogic(shape) { 
     var positive: Long = 0 
     var negative: Long = 0 
     setHandler(in, new InHandler { 
     override def onPush(): Unit = { 
      val elem = grab(in) 
      if (p(elem)) { 
      push(out, elem) 
      positive += 1 
      } else { 
      pull(in) 
      negative += 1 
      } 
     } 
     override def onUpstreamFinish(): Unit = { 
      result.success((positive,negative)) 
      completeStage() 
     } 
     }) 
     setHandler(out, new OutHandler { 
     override def onPull(): Unit = { 
      pull(in) 
     } 
     }) 
     (logic, result.future) 
    } 
}