Allow overriding dispatcher in mapWithResource (#1949)

* Allow overriding the dispatcher in mapWithResource

Closes #1948
This commit is contained in:
Piotr Sowiński 2025-07-15 06:21:24 +02:00 committed by GitHub
parent 5ca1b3846e
commit 6ef6ae1446
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 27 additions and 2 deletions

View file

@ -33,7 +33,7 @@ import com.google.common.jimfs.{ Configuration, Jimfs }
import org.apache.pekko
import pekko.Done
import pekko.stream.{ AbruptTerminationException, ActorAttributes, ActorMaterializer, SystemMaterializer }
import pekko.stream.{ AbruptTerminationException, ActorAttributes, ActorMaterializer, Attributes, SystemMaterializer }
import pekko.stream.ActorAttributes.supervisionStrategy
import pekko.stream.Supervision.{ restartingDecider, resumingDecider, stoppingDecider }
import pekko.stream.impl.{ PhasedFusingActorMaterializer, StreamSupervisor }
@ -231,6 +231,31 @@ class FlowMapWithResourceSpec extends StreamSpec(UnboundedMailboxConfig) {
finally p.cancel()
}
"allow overriding the default dispatcher" in {
val p = Source
.single(1)
.mapWithResource(() => newBufferedReader())(
(reader, _) => Option(reader.readLine()),
reader => {
reader.close()
None
})
.withAttributes(
Attributes.name("mapWithResourceCustomDispatcher") and
ActorAttributes.dispatcher("pekko.actor.default-dispatcher")
)
.runWith(TestSink.probe)
SystemMaterializer(system).materializer
.asInstanceOf[PhasedFusingActorMaterializer]
.supervisor
.tell(StreamSupervisor.GetChildren, testActor)
val ref = expectMsgType[Children].children
.find(_.path.toString contains "mapWithResourceCustomDispatcher").get
try assertDispatcher(ref, "pekko.actor.default-dispatcher")
finally p.cancel()
}
"fail when create throws exception" in {
EventFilter[TE](occurrences = 1).intercept {
val p = Source

View file

@ -1190,7 +1190,7 @@ trait FlowOps[+Out, +Mat] {
create,
(resource, out) => (resource, f(resource, out)),
resource => close(resource))
.withAttributes(DefaultAttributes.mapWithResource))
).withAttributes(DefaultAttributes.mapWithResource)
/**
* Transform each stream element with the help of an [[AutoCloseable]] resource and close it when the stream finishes or fails.