0

我一直在玩 Akka-Streams，我正在尝试Flow通过实现我自己的PushPullStage. 我希望Flow将从上游接收到的对象累积到一个列表中，并在上游完成时根据某些功能将它们分组，然后再将组发送到下游。

实现起来似乎很简单，但我不知道该怎么做！似乎没有办法从PushPullStage.

到目前为止，这是我的实现：

class Accumulate[A] extends PushPullStage[A, List[A]] {
    private var groups: List[List[A]] = Nil

    private def group(x: A): List[List[A]] = ...

    override def onPush(elem: A, ctx: Context[A]): SyncDirective = {
      groups = group(elem)
      ctx.pull()
    }

    override def onPull(ctx: Context[A]): SyncDirective =
      if (ctx.isFinishing) {
        for(group <- groups)
          ctx.push(group)    // this doesn't work

        ctx.finish()
      } else {
        ctx.pull()
      }

    override def onUpstreamFinish(ctx: Context[A]): TerminationDirective =
      ctx.absorbTermination()
  }
}

编辑

我更改了代码以考虑背压，现在一切正常。基本上我只需要让下游Flow做他们应该做的事情并继续拉动元素：

class Accumulate[A] extends PushPullStage[A, List[A]] {
    private var groups: List[List[A]] = Nil

    private def group(x: A): List[List[A]] = ...

    override def onPush(elem: A, ctx: Context[A]): SyncDirective = {
      groups = group(elem)
      ctx.pull()
    }

    override def onPull(ctx: Context[A]): SyncDirective =
      if (ctx.isFinishing) {
        groups match {
          case Nil => ctx.finish()

          case head :: tail =>
            groups = tail
            ctx.push(head)
        }
      } else {
        ctx.pull()
      }

    override def onUpstreamFinish(ctx: Context[A]): TerminationDirective =
      ctx.absorbTermination()
  }
}
4

1 回答 1

2

你不能推得比要求的更多，因为那会违反背压。另外，值得注意的是，我不建议您尝试执行此操作，因为对于大型或无界流，这会导致 OutOfMemoryError 爆炸。

class Accumulate[A] extends PushPullStage[A, List[A]] {
    private var groups: List[List[A]] = Nil

    private def group(x: A): List[List[A]] = ...

    override def onPush(elem: A, ctx: Context[A]): SyncDirective = {
      groups = group(elem)
      ctx.pull()
    }

    override def onPull(ctx: Context[A]): SyncDirective =
      if (ctx.isFinishing) {
        groups match {
          case Nil => ctx.finish()
          case group :: rest =>
            groups = rest
            ctx.push(group)
        }
      } else {
        ctx.pull()
      }

    override def onUpstreamFinish(ctx: Context[A]): TerminationDirective =
      ctx.absorbTermination()
  }
}
于 2015-09-19T21:25:16.033 回答