This is an automated email from the ASF dual-hosted git repository.
pjfanning 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 b1ff0db36c feat: support default attributes on materializers (#3647)
b1ff0db36c is described below
commit b1ff0db36c44f37f4e5c5b02bf3e4742090dac55
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 11:02:33 2026 +0100
feat: support default attributes on materializers (#3647)
* feat: support default attributes on materializers
Motivation:
There is no way to create a materializer that applies a set of default
attributes (for example a dispatcher, buffer sizes or a supervision
strategy) to every stream it materializes; the attributes have to be put
on each stream individually, or changed globally via configuration.
Modification:
- Add `Materializer(contextProvider, defaultAttributes)` /
`Materializer.createMaterializer(contextProvider, defaultAttributes)`
for actor-context bound materializers.
- Add `Materializer(systemProvider, defaultAttributes)` /
`Materializer.createMaterializer(systemProvider, defaultAttributes)`
for system level materializers, backed by a new internal
`SystemMaterializer.createAdditionalSystemMaterializer(Attributes)`.
- Add a new internal `MaterializerGuardian.StartMaterializerWithAttributes`
message; the existing `StartMaterializer` case object is unchanged.
- The given attributes are combined with the configured materializer
settings attributes, so attributes set on a stream still take precedence.
- Fix the Scaladoc of the Java `createMaterializer(systemProvider)` and
mention the new option in the stream basics docs.
Result:
Users can create materializers whose default attributes apply to all
streams they materialize. All changes are additive, no MiMa excludes.
Tests:
- Not run locally - left to CI; scalafmt applied,
stream-tests/Test/headerCreateAll run for the new spec
References:
None - new materializer factory overloads
* test: run MaterializerWithAttributesSpec actor on the stream test
dispatcher
Motivation:
The actor context materializer test created its actor on the default
dispatcher. In stream tests the default mailbox rejects actors from the
org.apache.pekko.stream package on the default dispatcher, so the actor
failed to start and the test timed out.
Modification:
Create the test actor with `withDispatcher("pekko.test.stream-dispatcher")`,
as other stream tests do.
Result:
The actor starts and the test can check the materializer attributes.
Tests:
- Not run locally - left to CI; scalafmt applied
References:
Refs #3647
* Retain Lightbend source notices for adapted materializer code
Motivation:
Review of #3647 noted that MaterializerWithAttributesSpec and the new
Materializer overloads are adapted from Akka 2.9.0 sources that carry 2023
Lightbend copyright notices, and that a docs line had trailing whitespace.
Modification:
- MaterializerWithAttributesSpec keeps its source notice
(Copyright (C) 2023 Lightbend Inc.), with the Apache header added by
headerCreate.
- Update the Materializer.scala notice from 2015-2022 to the source notice
2015-2023.
- Remove the trailing space in stream-flows-and-basics.md.
Result:
Source notices of the adapted code are preserved, and git diff --check
passes.
Tests:
sbt "stream-tests/Test/headerCreate" "stream/headerCheckAll"
"stream-tests/headerCheckAll"
sbt "stream-tests/testOnly
org.apache.pekko.stream.MaterializerWithAttributesSpec" (3 passed)
git diff --check
References:
https://github.com/apache/pekko/pull/3647
---
.../main/paradox/stream/stream-flows-and-basics.md | 9 +-
.../stream/MaterializerWithAttributesSpec.scala | 115 +++++++++++++++++++++
.../org/apache/pekko/stream/Materializer.scala | 62 ++++++++++-
.../apache/pekko/stream/SystemMaterializer.scala | 10 ++
.../pekko/stream/impl/MaterializerGuardian.scala | 7 ++
5 files changed, 198 insertions(+), 5 deletions(-)
diff --git a/docs/src/main/paradox/stream/stream-flows-and-basics.md
b/docs/src/main/paradox/stream/stream-flows-and-basics.md
index 2bb05d8994..71c25c0ef5 100644
--- a/docs/src/main/paradox/stream/stream-flows-and-basics.md
+++ b/docs/src/main/paradox/stream/stream-flows-and-basics.md
@@ -271,8 +271,11 @@ well-known sinks, such as @scala[`runForeach(el =>
...)`]@java[`runForeach(el ->
Materialization is performed synchronously on the materializing thread by an
@apidoc[actor.ActorSystem] global @apidoc[stream.Materializer].
The actual stream processing is handled by actors started up during the
streams materialization,
-which will be running on the thread pools they have been configured to run on
- which defaults to the dispatcher set in
-the `ActorSystem` config or provided as attributes on the stream that is
getting materialized.
+which will be running on the thread pools they have been configured to run on.
By default, the thread pool used is the
+dispatcher set in the `ActorSystem` config, but it is possible to select other
thread pools by providing @apidoc[stream.Attributes] to
+
+ * the stream that is getting materialized
+ * a custom instance of @apidoc[stream.Materializer] created with
`defaultAttributes`
@@@ note
@@ -384,7 +387,7 @@ and emitting the "materialized value". An
@apidoc[actor.ActorSystem] wide `Mater
@apidoc[SystemMaterializer] by @scala[having an implicit `ActorSystem` in
scope]@java[passing the `ActorSystem` to the
various `run` methods] this way there is no need to worry about the
`Materializer` unless there are special requirements.
-The use case that may require a custom instance of `Materializer` is when all
streams materialized in an actor should be tied to the Actor lifecycle and stop
if the Actor stops or crashes.
+One use case that may require a custom instance of `Materializer` is when all
streams materialized in an actor should be tied to the Actor lifecycle and stop
if the Actor stops or crashes.
An important aspect of working with streams and actors is understanding a
`Materializer`'s life-cycle.
The materializer is bound to the lifecycle of the
@apidoc[actor.ActorRefFactory] it is created from, which in practice will
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/MaterializerWithAttributesSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/MaterializerWithAttributesSpec.scala
new file mode 100644
index 0000000000..f33729dc95
--- /dev/null
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/MaterializerWithAttributesSpec.scala
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.stream
+
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.Actor
+import pekko.actor.Props
+import pekko.stream.scaladsl.Sink
+import pekko.stream.scaladsl.Source
+import pekko.stream.stage.GraphStageLogic
+import pekko.stream.stage.GraphStageWithMaterializedValue
+import pekko.stream.stage.OutHandler
+import pekko.testkit.ImplicitSender
+import pekko.testkit.PekkoSpec
+
+object MaterializerWithAttributesSpec {
+
+ /** Completes immediately and materializes the attributes it was
materialized with */
+ object AttributesSource extends
GraphStageWithMaterializedValue[SourceShape[Nothing], Attributes] {
+ val out: Outlet[Nothing] = Outlet[Nothing]("AttributesSource.out")
+ override val shape: SourceShape[Nothing] = SourceShape(out)
+
+ override def createLogicAndMaterializedValue(inheritedAttributes:
Attributes): (GraphStageLogic, Attributes) = {
+ val logic = new GraphStageLogic(shape) with OutHandler {
+ override def preStart(): Unit = completeStage()
+ override def onPull(): Unit = completeStage()
+
+ setHandler(out, this)
+ }
+ (logic, inheritedAttributes)
+ }
+ }
+
+ class StreamActor extends Actor {
+ override def receive: Receive = {
+ case "do it!" =>
+ val materializerWithAttributes: Materializer = Materializer(context,
Attributes.name("bar"))
+
+ val firstAttributes =
+
Source.fromGraph(AttributesSource).to(Sink.ignore).run()(materializerWithAttributes)
+ val secondAttributes =
+
Source.fromGraph(AttributesSource).to(Sink.ignore).run()(Materializer(context))
+
+ sender() ! firstAttributes
+ sender() ! secondAttributes
+
+ context.stop(self)
+ }
+ }
+}
+
+class MaterializerWithAttributesSpec extends PekkoSpec with ImplicitSender {
+ import MaterializerWithAttributesSpec._
+
+ private def names(attributes: Attributes): List[String] =
+ attributes.attributeList.collect { case Attributes.Name(n) => n }
+
+ "A system level materializer with default attributes" should {
+ "apply its attributes to streams it materializes" in {
+ val materializerWithAttributes = Materializer(system,
Attributes.name("foo"))
+
+ val attributes =
Source.fromGraph(AttributesSource).to(Sink.ignore).run()(materializerWithAttributes)
+ names(attributes) should contain("foo")
+
+ val otherAttributes =
Source.fromGraph(AttributesSource).to(Sink.ignore).run()(Materializer(system))
+ names(otherAttributes) shouldNot contain("foo")
+
+ materializerWithAttributes.shutdown()
+ }
+
+ "let attributes on the stream override the materializer default
attributes" in {
+ val materializerWithAttributes = Materializer(system,
Attributes.inputBuffer(1, 1))
+
+ val attributes =
Source.fromGraph(AttributesSource).to(Sink.ignore).run()(materializerWithAttributes)
+ attributes.get[Attributes.InputBuffer] should
===(Some(Attributes.InputBuffer(1, 1)))
+
+ val overridden = Source
+ .fromGraph(AttributesSource)
+ .withAttributes(Attributes.inputBuffer(2, 2))
+ .to(Sink.ignore)
+ .run()(materializerWithAttributes)
+ overridden.get[Attributes.InputBuffer] should
===(Some(Attributes.InputBuffer(2, 2)))
+
+ materializerWithAttributes.shutdown()
+ }
+ }
+
+ "An actor context materializer with default attributes" should {
+ "apply its attributes to streams it materializes" in {
+ val streamActor = system.actorOf(Props(new
StreamActor).withDispatcher("pekko.test.stream-dispatcher"))
+
+ streamActor ! "do it!"
+
+ val responses = receiveN(2, 3.seconds)
+ responses.filterNot(_.isInstanceOf[Attributes]) shouldBe empty
+
+ val allAttributes = responses.map(_.asInstanceOf[Attributes])
+ names(allAttributes.head) should contain("bar")
+ names(allAttributes.last) shouldNot contain("bar")
+ }
+ }
+}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/Materializer.scala
b/stream/src/main/scala/org/apache/pekko/stream/Materializer.scala
index dd5c56b92b..d9e29de278 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/Materializer.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/Materializer.scala
@@ -8,7 +8,7 @@
*/
/*
- * Copyright (C) 2015-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2015-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.stream
@@ -27,6 +27,7 @@ import pekko.actor.Props
import pekko.annotation.DoNotInherit
import pekko.annotation.InternalApi
import pekko.event.LoggingAdapter
+import pekko.stream.impl.PhasedFusingActorMaterializer
/**
* The Materializer is the component responsible for turning a stream
blueprint into a running stream.
@@ -212,6 +213,37 @@ object Materializer {
*/
def createMaterializer(contextProvider: ClassicActorContextProvider):
Materializer = apply(contextProvider)
+ /**
+ * Scala API: Create a materializer whose lifecycle will be tied to the one
of the passed actor context.
+ * When the actor stops the materializer will stop and all streams created
with it will be failed with an [[AbruptTerminationExeption]]
+ *
+ * The given `defaultAttributes` are applied to all streams materialized
with this materializer, and can be
+ * overridden by attributes set on the streams themselves.
+ *
+ * You can pass either a classic actor context or a typed actor context.
+ *
+ * @since 2.0.0
+ */
+ def apply(contextProvider: ClassicActorContextProvider, defaultAttributes:
Attributes): Materializer = {
+ val context = contextProvider.classicActorContext
+ val settings = SystemMaterializer(context.system).materializerSettings
+ PhasedFusingActorMaterializer(context, "flow", settings,
settings.toAttributes.and(defaultAttributes))
+ }
+
+ /**
+ * Java API: Create a materializer whose lifecycle will be tied to the one
of the passed actor context.
+ * When the actor stops the materializer will stop and all streams created
with it will be failed with an [[AbruptTerminationExeption]]
+ *
+ * The given `defaultAttributes` are applied to all streams materialized
with this materializer, and can be
+ * overridden by attributes set on the streams themselves.
+ *
+ * You can pass either a classic actor context or a typed actor context.
+ *
+ * @since 2.0.0
+ */
+ def createMaterializer(contextProvider: ClassicActorContextProvider,
defaultAttributes: Attributes): Materializer =
+ apply(contextProvider, defaultAttributes)
+
/**
* Scala API: Create a new materializer that will stay alive as long as the
system does or until it is explicitly stopped.
*
@@ -224,7 +256,7 @@ object Materializer {
SystemMaterializer(systemProvider.classicSystem).createAdditionalSystemMaterializer()
/**
- * Scala API: Create a new materializer that will stay alive as long as the
system does or until it is explicitly stopped.
+ * Java API: Create a new materializer that will stay alive as long as the
system does or until it is explicitly stopped.
*
* *Note* prefer using the default [[SystemMaterializer]] by passing the
`ActorSystem` to the various `run`
* methods on the streams. Only create new system level materializers if you
have specific
@@ -234,6 +266,32 @@ object Materializer {
def createMaterializer(systemProvider: ClassicActorSystemProvider):
Materializer =
apply(systemProvider)
+ /**
+ * Scala API: Create a new materializer that will stay alive as long as the
system does or until it is explicitly stopped.
+ *
+ * The given `defaultAttributes` are applied to all streams materialized
with this materializer, and can be
+ * overridden by attributes set on the streams themselves.
+ *
+ * It is generally advised to limit the number of system level materializers
created.
+ *
+ * @since 2.0.0
+ */
+ def apply(systemProvider: ClassicActorSystemProvider, defaultAttributes:
Attributes): Materializer =
+
SystemMaterializer(systemProvider.classicSystem).createAdditionalSystemMaterializer(defaultAttributes)
+
+ /**
+ * Java API: Create a new materializer that will stay alive as long as the
system does or until it is explicitly stopped.
+ *
+ * The given `defaultAttributes` are applied to all streams materialized
with this materializer, and can be
+ * overridden by attributes set on the streams themselves.
+ *
+ * It is generally advised to limit the number of system level materializers
created.
+ *
+ * @since 2.0.0
+ */
+ def createMaterializer(systemProvider: ClassicActorSystemProvider,
defaultAttributes: Attributes): Materializer =
+ apply(systemProvider, defaultAttributes)
+
}
/**
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/SystemMaterializer.scala
b/stream/src/main/scala/org/apache/pekko/stream/SystemMaterializer.scala
index b388de38bb..412a470101 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/SystemMaterializer.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/SystemMaterializer.scala
@@ -80,6 +80,16 @@ final class SystemMaterializer(system: ExtendedActorSystem)
extends Extension {
Await.result(started, materializerTimeout.duration).materializer
}
+ /**
+ * INTERNAL API
+ */
+ @InternalApi
+ private[pekko] def createAdditionalSystemMaterializer(defaultAttributes:
Attributes): Materializer = {
+ val started = (materializerGuardian ?
MaterializerGuardian.StartMaterializerWithAttributes(defaultAttributes))
+ .mapTo[MaterializerGuardian.MaterializerStarted]
+ Await.result(started, materializerTimeout.duration).materializer
+ }
+
/**
* INTERNAL API
*/
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/MaterializerGuardian.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/MaterializerGuardian.scala
index c200bbf651..bb76d2f105 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/MaterializerGuardian.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/MaterializerGuardian.scala
@@ -21,6 +21,7 @@ import pekko.actor.Actor
import pekko.actor.Props
import pekko.annotation.InternalApi
import pekko.stream.ActorMaterializerSettings
+import pekko.stream.Attributes
import pekko.stream.Materializer
/**
@@ -33,6 +34,7 @@ import pekko.stream.Materializer
private[pekko] object MaterializerGuardian {
case object StartMaterializer
+ final case class StartMaterializerWithAttributes(attributes: Attributes)
final case class MaterializerStarted(materializer: Materializer)
// this is available to keep backwards compatibility with ActorMaterializer
and should
@@ -63,6 +65,11 @@ private[pekko] final class MaterializerGuardian(
override def receive: Receive = {
case StartMaterializer =>
sender() ! MaterializerStarted(startMaterializer(defaultNamePrefix,
None))
+ case StartMaterializerWithAttributes(attributes) =>
+ val materializer =
+ PhasedFusingActorMaterializer(context, defaultNamePrefix,
materializerSettings,
+ defaultAttributes.and(attributes))
+ sender() ! MaterializerStarted(materializer)
case LegacyStartMaterializer(namePrefix, settings) =>
sender() ! MaterializerStarted(startMaterializer(namePrefix,
Some(settings)))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
