Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add overridden duration timeout to StreamTestKit #1468

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add overridden duration timeout to StreamTestKit #1468

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ...lters/1.1.1.backwards.excludes/pr-1468-override-timeout-stream-testkit.backwards.excludes
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.testkit.scaladsl.StreamTestKit.assertNoChildren")
20 changes: 19 additions & 1 deletion stream-testkit/src/main/scala/org/apache/pekko/stream/testkit/javadsl/StreamTestKit.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,10 @@ import pekko.stream.{ Materializer, SystemMaterializer }
import pekko.stream.impl.PhasedFusingActorMaterializer
import pekko.stream.testkit.scaladsl

import java.time.Duration
import java.util.concurrent.TimeUnit
import scala.concurrent.duration.FiniteDuration

object StreamTestKit {

/**
Expand All @@ -29,7 +33,21 @@ object StreamTestKit {
def assertAllStagesStopped(mat: Materializer): Unit =
mat match {
case impl: PhasedFusingActorMaterializer =>
scaladsl.StreamTestKit.assertNoChildren(impl.system, impl.supervisor)
scaladsl.StreamTestKit.assertNoChildren(impl.system, impl.supervisor, None)
case _ =>
}

/**
* Assert that there are no stages running under a given materializer.
* Usually this assertion is run after a test-case to check that all of the
* stages have terminated successfully with an overridden duration that ignores
* `stream.testkit.all-stages-stopped-timeout`.
*/
def assertAllStagesStopped(mat: Materializer, overrideTimeout: Duration): Unit =
mat match {
case impl: PhasedFusingActorMaterializer =>
scaladsl.StreamTestKit.assertNoChildren(impl.system, impl.supervisor,
Some(FiniteDuration(overrideTimeout.toMillis, TimeUnit.MILLISECONDS)))
case _ =>
}

Expand Down
31 changes: 27 additions & 4 deletions stream-testkit/src/main/scala/org/apache/pekko/stream/testkit/scaladsl/StreamTestKit.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,12 +35,30 @@ object StreamTestKit {
* This assertion is useful to check that all of the stages have
* terminated successfully.
*/
def assertAllStagesStopped[T](block: => T, overrideTimeout: FiniteDuration)(implicit materializer: Materializer): T =
materializer match {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is code duplication here, I can clean it up if the premise of this PR is acceptable.

case impl: PhasedFusingActorMaterializer =>
stopAllChildren(impl.system, impl.supervisor)
val result = block
assertNoChildren(impl.system, impl.supervisor, Some(overrideTimeout))
result
case _ => block
}

/**
* Asserts that after the given code block is ran, no stages are left over
* that were created by the given materializer with an overridden duration
* that ignores `stream.testkit.all-stages-stopped-timeout`.
*
* This assertion is useful to check that all of the stages have
* terminated successfully.
*/
def assertAllStagesStopped[T](block: => T)(implicit materializer: Materializer): T =
materializer match {
case impl: PhasedFusingActorMaterializer =>
stopAllChildren(impl.system, impl.supervisor)
val result = block
assertNoChildren(impl.system, impl.supervisor)
assertNoChildren(impl.system, impl.supervisor, None)
result
case _ => block
}
Expand All @@ -53,10 +71,15 @@ object StreamTestKit {
}

/** INTERNAL API */
@InternalApi private[testkit] def assertNoChildren(sys: ActorSystem, supervisor: ActorRef): Unit = {
@InternalApi private[testkit] def assertNoChildren(sys: ActorSystem, supervisor: ActorRef,
overrideTimeout: Option[FiniteDuration]): Unit = {
val probe = TestProbe()(sys)
val c = sys.settings.config.getConfig("pekko.stream.testkit")
val timeout = c.getDuration("all-stages-stopped-timeout", MILLISECONDS).millis

val timeout = overrideTimeout.getOrElse {
val c = sys.settings.config.getConfig("pekko.stream.testkit")
c.getDuration("all-stages-stopped-timeout", MILLISECONDS).millis
}

probe.within(timeout) {
try probe.awaitAssert {
supervisor.tell(StreamSupervisor.GetChildren, probe.ref)
Expand Down
40 changes: 40 additions & 0 deletions stream-testkit/src/test/scala/org/apache/pekko/stream/testkit/StreamConfiguration.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.pekko.stream.testkit

import org.apache.pekko.testkit.TestKitBase
import org.scalatest.time.{ Millis, Span }

import java.util.concurrent.TimeUnit

trait StreamConfiguration extends TestKitBase {
case class StreamConfig(allStagesStoppedTimeout: Span = Span({
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The premise here is that by default we get the timeout value from Pekko typesafe config (as is expected when using Pekko) however if a user wants to override this then its done the ScalaTest idiomatic way (which is overriding a def in a trait, similar to how PatienceConfig works).

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is possible to omit units and support more flexibility, but the code is slightly complex and for reference only.

import scala.jdk.DurationConverters._
import org.scalatest.time.Span._
system.settings.config.getDuration("pekko.stream.testkit.all-stages-stopped-timeout").toScala.convertDurationToSpan

I just saw this {} and wanted to make some changes. I looked at the Span code and found that it has convertDurationoSpan. You can do whatever you want.

val c = system.settings.config.getConfig("pekko.stream.testkit")
c.getDuration("all-stages-stopped-timeout", TimeUnit.MILLISECONDS)
}, Millis))

private val defaultStreamConfig = StreamConfig()

/**
* The default `StreamConfig` which is derived from the Actor System's `pekko.stream.testkit.all-stages-stopped-timeout`
* configuration value. If you want to provide a different StreamConfig for specific tests without having to re-specify
* `pekko.stream.testkit.all-stages-stopped-timeout` then you can override this value.
*/
implicit def streamConfig: StreamConfig = defaultStreamConfig

}
8 changes: 6 additions & 2 deletions stream-testkit/src/test/scala/org/apache/pekko/stream/testkit/StreamSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,10 @@ import org.scalatest.Failed

import com.typesafe.config.{ Config, ConfigFactory }

abstract class StreamSpec(_system: ActorSystem) extends PekkoSpec(_system) {
import java.util.concurrent.TimeUnit

abstract class StreamSpec(_system: ActorSystem) extends PekkoSpec(_system) with StreamConfiguration {

def this(config: Config) =
this(
ActorSystem(
Expand Down Expand Up @@ -73,7 +76,8 @@ abstract class StreamSpec(_system: ActorSystem) extends PekkoSpec(_system) {
case impl: PhasedFusingActorMaterializer =>
stopAllChildren(impl.system, impl.supervisor)
val result = test.apply()
assertNoChildren(impl.system, impl.supervisor)
assertNoChildren(impl.system, impl.supervisor,
Some(FiniteDuration(streamConfig.allStagesStoppedTimeout.millisPart, TimeUnit.MILLISECONDS)))
result
case _ => other
}
Expand Down
Loading