我应该如何测试 akka-streams 重新启动源代码的使用情况

Question

我正在开发一个应用程序，该应用程序有几个长时间运行的流，它订阅有关某个实体的数据并处理该数据。这些流应该 24/7 全天候运行，因此我们需要处理故障(网络问题等(。

为此，我们将我们的来源包装在RestartingSource.

我现在正在尝试验证这种行为，虽然它看起来可以正常工作，但我正在努力创建一个测试，在其中我推送一些数据，验证它是否正确处理，然后发送错误，并验证它是否在那之后重新连接并继续处理。

我将其归结为这个最小案例：

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{RestartSource, Sink, Source}
import akka.stream.testkit.TestPublisher
import org.scalatest.concurrent.Eventually
import org.scalatest.{FlatSpec, Matchers}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext
class MinimalSpec extends FlatSpec with Matchers with Eventually {
"restarting a failed source" should "be testable" in {
implicit val sys: ActorSystem = ActorSystem("akka-grpc-measurements-for-test")
implicit val mat: ActorMaterializer = ActorMaterializer()
implicit val ec: ExecutionContext = sys.dispatcher
val probe = TestPublisher.probe[Int]()
val restartingSource = RestartSource
.onFailuresWithBackoff(1 second, 1 minute, 0d) { () => Source.fromPublisher(probe) }
var last: Int = 0
val sink = Sink.foreach { l: Int => last = l }
restartingSource.runWith(sink)
probe.sendNext(1)
eventually {
last shouldBe 1
}
probe.sendNext(2)
eventually {
last shouldBe 2
}
probe.sendError(new RuntimeException("boom"))
probe.expectSubscription()
probe.sendNext(3)
eventually {
last shouldBe 3
}
}
}

此测试在最后一个eventually块上始终失败，并带有Last failure message: 2 was not equal to 3。我在这里错过了什么？

编辑：阿卡版本2.5.31

Answer 1

我在看了TestPublisher代码后想通了。它的订阅是一个lazy val。因此，当RestartSource检测到错误并再次() => Source.fromPublisher(probe)执行工厂方法时，它会得到一个新的Source，但probe的subscription仍然指向旧的Source。更改代码以初始化新Source和TestPublisher工作。