Akka流，源于函数？

Question

我想要一个以给定的时间间隔计算函数并发出其输出的源。作为一种变通方法，我可以通过源代码来实现。排队提供，但还没有找到更干净的方法。理想情况下，我会有

def myFunction() = ....                     // function with side-effects 
Source.tick(1.second, 1.second, myFunction) // myFunction is evaluated at every tick

有什么想法吗？

Answer 1

我想，油门就是你需要的。完全可运行的示例，将Source应用于iterable，它使用next（）中的函数：

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.ThrottleMode.Shaping
import akka.stream.scaladsl.Source

import scala.concurrent.duration._

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
var i = 0

def myFunction(): Int = {
  i = i + 1
  i
}

import scala.collection.immutable.Iterable

val x: Iterable[Int] = new Iterable[Int] {
  override def iterator: Iterator[Int] =
    new Iterator[Int] {
      override def hasNext: Boolean = true

      override def next(): Int = myFunction()
    }
}
Source(x).throttle(1, 1.second, 1, Shaping).runForeach(println)

throttle参数：节流源，每1秒1个元素，最大突发=1，在发出消息之前暂停，以满足节流速率（这就是成形）。

Answer 2

最干净的方法可能是使用map

Source.tick(1.second, 1.second, NotUsed).map(_ ⇒ myFunction())