scala - 如何结束无限的akka​​流

Question

我是 Akka Streams 的新手，但有一个案例我想用它来寻找来自无限来源的排列。具有有限源的简化示例可能如下所示。

val future = Source(1 to 100)
    .map { i => if (i % 20 == 0) println(i); i }
    .filter(_ == 42)
    .runWith(Sink.fold[Int, Int](0)(Keep.right))

此示例输出：

20
40
60
80
100

我显然对源过去很好，42但我不想在能够得到结果之前耗尽整个流。

val result: Int = Await.result(future, 1.second)
result should be(42)

问题是，当我找到我要找的东西时，我应该如何结束流？

Answer 1

val future = Source(1 to 100)
    .map { i => if (i % 20 == 0) println(i); i }
    .filter(_ == 42)
    .runWith(Sink.head)

Answer 2

推广到N值，例如大于或等于 42 的 10 个值，您可以使用grouped：

val N = 10

val future : Future[Seq[Int]] = 
  Source(1 to 100).map { i => if (i % 20 == 0) println(i); i }
                  .filter(_ >= 42)
                  .grouped(N)
                  .runWith(Sink.head)