如何从Source[A]创建Akka流源[Seq[A]]

Question

在Akka Streams的早期版本中，groupBy返回了一个Source的Source，可以具体化为源[Seq[a]]。

在Akka Streams 2.4中，我看到groupBy返回一个子流——我不清楚如何使用它。我需要应用于流的转换必须使整个Seq可用，所以我不能只映射子流（我认为）。

我已经编写了一个扩展GraphStage的类，该类通过GraphStageLogic中的可变集合进行聚合，但是否有内置功能？我是否错过了SubFlow的要点？




Answer 1

最后我写了一个图形舞台：

class FlowAggregation[A, B](f: A => B) extends GraphStage[FlowShape[A, Seq[A]]] {
  val in: Inlet[A] = Inlet("in")
  val out: Outlet[Seq[A]] = Outlet("out")
  override val shape = FlowShape.of(in, out)

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) {

      private var counter: Option[B] = None
      private var aggregate = scala.collection.mutable.ArrayBuffer.empty[A]

      setHandler(in, new InHandler {
        override def onPush(): Unit = {
          val element = grab(in)

          counter.fold({
            counter = Some(f(element))
            aggregate += element
            pull(in)
          }) { p =>
            if (f(element) == p) {
              aggregate += element
              pull(in)
            } else {
              push(out, aggregate)
              counter = Some(f(element))
              aggregate = scala.collection.mutable.ArrayBuffer(element)
            }
          }
        }
        override def onUpstreamFinish(): Unit = {
          emit(out, aggregate)
          complete(out)
        }
      })

      setHandler(out, new OutHandler {
        override def onPull(): Unit = {
          pull(in)
        }
      })
    }
}