def run()

in testkit/src/main/scala/org/apache/pekko/projection/testkit/scaladsl/ProjectionTestKit.scala [57:116]

• 24 lines of code
• 1 McCabe index (conditional complexity)

  def run(projection: Projection[_])(assertFunction: => Unit): Unit =
    runInternal(projection, assertFunction, testKit.testKitSettings.SingleExpectDefaultTimeout, 100.millis)

  /**
   * Run a Projection and assert its projected data using the passed assert function and the max duration of the test.
   *
   * Projection is started and stopped by the TestKit. While the projection is running, the assert function
   * will be called every 100 milliseconds until it completes without errors (no exceptions or assertion errors are thrown).
   *
   * If the assert function doesn't complete without error within the passed `max` duration the test will fail.
   *
   * Note: when testing a Projection with this method, the Restart Backoff is disabled.
   * Any backoff configuration settings from `.conf` file or programmatically added will be overwritten.
   *
   * @param projection - the Projection to run
   * @param max - FiniteDuration delimiting the max duration of the test
   * @param assertFunction - a by-name code block that exercise the test assertions
   */
  def run(projection: Projection[_], max: FiniteDuration)(assertFunction: => Unit): Unit =
    runInternal(projection, assertFunction, max, 100.millis)

  /**
   * Run a Projection and assert its projected data using the passed assert function,
   * the max duration of the test and the interval between each assertion.
   *
   * Projection is started and stopped by the TestKit. While the projection is running, the assert function
   * will be called every `interval` until it completes without errors (no exceptions or assertion errors are thrown).
   *
   * If the assert function doesn't complete without error within the passed `max` duration the test will fail.
   *
   * Note: when testing a Projection with this method, the Restart Backoff is disabled.
   * Any backoff configuration settings from `.conf` file or programmatically added will be overwritten.
   *
   * @param projection - the Projection to run
   * @param max - FiniteDuration delimiting the max duration of the test
   * @param interval - FiniteDuration defining the interval in each the assert function will be called
   * @param assertFunction - a by-name code block that exercise the test assertions
   */
  def run(projection: Projection[_], max: FiniteDuration, interval: FiniteDuration)(assertFunction: => Unit): Unit =
    runInternal(projection, assertFunction, max, interval)

  private def runInternal(
      projection: Projection[_],
      assertFunction: => Unit,
      max: FiniteDuration,
      interval: FiniteDuration): Unit = {

    val probe = testKit.createTestProbe[Nothing]("internal-projection-testkit-probe")
    val actorHandler = spawnActorHandler(projection)
    val running =
      projection
        .withRestartBackoff(0.millis, 0.millis, 0.0, 0)
        .run()(testKit.system)
    try {
      probe.awaitAssert(assertFunction, max, interval)
    } finally {
      Await.result(running.stop(), max)
      actorHandler.foreach(ref => testKit.stop(ref))
    }
  }