我有一个无限的元素流,我想按id和聚合组进行分组,让我们说2秒,然后将它们发送到下游.这是一个不起作用的代码,但可以更好地解释我想要的东西：
Source
.tick(0 second,50 millis,() => if (Random.nextBoolean) (1,s"A") else (2,s"B"))
.map { f => f() }
.groupBy(10,_._1)
// how to aggregate grouped elements here for two seconds?
.scan(Seq[String]()) { (x,y) => x ++ Seq(y._2) }
.to(Sink.foreach(println))
期望的输出应该如下所示：
Seq(A,A,A) Seq(B,B,B) Seq(A,B) // and so on
如何使用流实现此类功能？
解决方法
@H_301_34@
