This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 0059ccec21 test: make mapWithResource cleanup tests deterministic
(#3319)
0059ccec21 is described below
commit 0059ccec214764ff3fd95290e9ff21cc0deca651
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sat Jul 11 06:50:21 2026 +0800
test: make mapWithResource cleanup tests deterministic (#3319)
Motivation:
The abrupt materializer termination tests used Source.single, which can
complete across mapWithResource's async boundary and close the resource before
shutdown.
Modification:
Keep the upstream open with Source.maybe and assert that the resource has
not been closed before shutting down the materializer.
Result:
The tests now distinguish abrupt termination cleanup from normal upstream
completion.
Tests:
- FlowMapWithResourceSpec on Scala 2.13 and Scala 3
- sbt +headerCheckAll
- sbt checkCodeStyle
- sbt validatePullRequest (3392 passed, 0 failed)
References:
Refs #3310
---
.../org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala
index 103d07ff09..967bbeff1d 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapWithResourceSpec.scala
@@ -416,7 +416,7 @@ class FlowMapWithResourceSpec extends
StreamSpec(UnboundedMailboxConfig) {
val promise = Promise[Done]()
val created = Promise[Done]()
val matVal = Source
- .single(1)
+ .maybe[Int]
.mapWithResource(() => {
created.trySuccess(Done)
newBufferedReader()
@@ -429,6 +429,7 @@ class FlowMapWithResourceSpec extends
StreamSpec(UnboundedMailboxConfig) {
.mapConcat(identity)
.runWith(Sink.never)(mat)
Await.result(created.future, 3.seconds.dilated) shouldBe Done
+ promise.isCompleted shouldBe false
mat.shutdown()
matVal.failed.futureValue shouldBe an[AbruptTerminationException]
Await.result(promise.future, 3.seconds.dilated) shouldBe Done
@@ -531,10 +532,11 @@ class FlowMapWithResourceSpec extends
StreamSpec(UnboundedMailboxConfig) {
}
}
val matVal = Source
- .single(1)
+ .maybe[Int]
.mapWithResource(create, (_: AutoCloseable, count) => count)
.runWith(Sink.never)(mat)
Await.result(created.future, 3.seconds.dilated) shouldBe Done
+ closedCounter.get shouldBe 0
mat.shutdown()
matVal.failed.futureValue shouldBe an[AbruptTerminationException]
Await.result(promise.future, 3.seconds.dilated) shouldBe Done
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
