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 535a5facc3 refactor: remove deprecated materializer settings APIs
(#3293)
535a5facc3 is described below
commit 535a5facc303eff7a897436cf37b85050d289e8c
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jul 5 23:13:46 2026 +0800
refactor: remove deprecated materializer settings APIs (#3293)
Motivation:
Several ActorMaterializerSettings builder methods and materializer-specific
settings types were deprecated since Akka 2.6.0. For Pekko 2.0.0 these public
APIs can be removed while preserving internal defaults used by streams, TCP,
and Artery.
Modification:
Remove ActorMaterializerSettings.withInputBuffer, public withDispatcher,
withIOSettings, withStreamRefSettings, IOSettings, StreamRefSettings, and
StreamRefSettingsImpl. Inline IO defaults as internal ActorMaterializerSettings
fields, replace stream-ref defaults with internal StreamRefDefaultSettings,
update TCP and stream-ref stages, update and prune MiMa filters, and migrate
affected stream tests to config or Attributes.inputBuffer.
Result:
Deprecated materializer settings APIs are gone from the public/source API,
internal stream/TCP defaults continue to work, and stream-tests no longer
depend on removed builder methods.
Tests:
- sbt "stream / Compile / compile" - success
- sbt "stream / mimaReportBinaryIssues" - success
- sbt "+stream / mimaReportBinaryIssues" - success
- sbt "stream-tests / Test / compile" - success
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.StreamRefsSpec
org.apache.pekko.stream.scaladsl.FlowSpec
org.apache.pekko.stream.scaladsl.AttributesSpec
org.apache.pekko.stream.scaladsl.FlowRecoverWithSpec
org.apache.pekko.stream.scaladsl.FlowSlidingSpec
org.apache.pekko.stream.scaladsl.FlowZipWithIndexSpec
org.apache.pekko.stream.scaladsl.TakeLastSinkSpec" - 156 tests passed, 1 pending
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.GraphFlowSpec" - 15 tests passed
- scalafmt --check --mode diff-ref=<merge-base> - success
- sbt headerCreateAll - success
- sbt "+headerCheckAll" - success
- git diff --check <merge-base> - success
- qodercli stdout review (/tmp/project-qoder-review-final.log) - No
must-fix findings
- subAgent review - No must-fix findings
- sbt sortImports - failed: Scalafix/scala.meta NoSuchMethodError in
multi-node-testkit and stream-tests; unrelated partial changes reverted
References:
Refs #3293
---
.../pekko/stream/scaladsl/AttributesSpec.scala | 5 +-
.../stream/scaladsl/FlowRecoverWithSpec.scala | 10 +-
.../pekko/stream/scaladsl/FlowSlidingSpec.scala | 11 ++-
.../apache/pekko/stream/scaladsl/FlowSpec.scala | 35 ++++---
.../stream/scaladsl/FlowZipWithIndexSpec.scala | 10 +-
.../stream/scaladsl/GraphBackedFlowSpec.scala | 10 +-
.../pekko/stream/scaladsl/TakeLastSinkSpec.scala | 10 +-
...emove-deprecated-materializer-settings.excludes | 32 ++++++
.../remove-deprecated-methods.excludes | 4 -
.../apache/pekko/stream/ActorMaterializer.scala | 110 ++++++---------------
.../apache/pekko/stream/StreamRefSettings.scala | 70 -------------
.../apache/pekko/stream/impl/io/TcpStages.scala | 6 +-
.../pekko/stream/impl/streamref/SinkRefImpl.scala | 12 +--
.../stream/impl/streamref/SourceRefImpl.scala | 12 +--
.../impl/streamref/StreamRefDefaultSettings.scala | 27 +++++
.../impl/streamref/StreamRefSettingsImpl.scala | 41 --------
16 files changed, 167 insertions(+), 238 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/AttributesSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/AttributesSpec.scala
index 84c090c4e3..cbb797d0ea 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/AttributesSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/AttributesSpec.scala
@@ -126,6 +126,9 @@ class AttributesSpec
extends StreamSpec(
ConfigFactory
.parseString("""
+ pekko.stream.materializer.initial-input-buffer-size = 2
+ pekko.stream.materializer.max-input-buffer-size = 16
+
my-dispatcher {
type = Dispatcher
executor = "thread-pool-executor"
@@ -141,7 +144,7 @@ class AttributesSpec
import AttributesSpec._
- val settings = ActorMaterializerSettings(system).withInputBuffer(initialSize
= 2, maxSize = 16)
+ val settings = ActorMaterializerSettings(system)
implicit val materializer: Materializer = ActorMaterializer(settings)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowRecoverWithSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowRecoverWithSpec.scala
index a0cafc337c..823340eed9 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowRecoverWithSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowRecoverWithSpec.scala
@@ -25,10 +25,16 @@ import pekko.stream.testkit.StreamSpec
import pekko.stream.testkit.Utils._
import pekko.stream.testkit.scaladsl.TestSink
+import com.typesafe.config.ConfigFactory
+
@nowarn // tests deprecated APIs
-class FlowRecoverWithSpec extends StreamSpec {
+class FlowRecoverWithSpec
+ extends StreamSpec(ConfigFactory.parseString("""
+ pekko.stream.materializer.initial-input-buffer-size = 1
+ pekko.stream.materializer.max-input-buffer-size = 1
+ """)) {
- val settings = ActorMaterializerSettings(system).withInputBuffer(initialSize
= 1, maxSize = 1)
+ val settings = ActorMaterializerSettings(system)
implicit val materializer: Materializer = ActorMaterializer(settings)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSlidingSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSlidingSpec.scala
index 141a550bd3..a1e10a8cae 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSlidingSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSlidingSpec.scala
@@ -23,10 +23,17 @@ import pekko.pattern.pipe
import pekko.stream.{ ActorMaterializer, ActorMaterializerSettings,
Materializer }
import pekko.stream.testkit._
+import com.typesafe.config.ConfigFactory
+
@nowarn
-class FlowSlidingSpec extends StreamSpec with ScalaCheckPropertyChecks {
+class FlowSlidingSpec
+ extends StreamSpec(ConfigFactory.parseString("""
+ pekko.stream.materializer.initial-input-buffer-size = 2
+ pekko.stream.materializer.max-input-buffer-size = 16
+ """))
+ with ScalaCheckPropertyChecks {
import system.dispatcher
- val settings = ActorMaterializerSettings(system).withInputBuffer(initialSize
= 2, maxSize = 16)
+ val settings = ActorMaterializerSettings(system)
implicit val materializer: Materializer = ActorMaterializer(settings)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSpec.scala
index eec0390ba1..c9f8f7cd33 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSpec.scala
@@ -48,16 +48,25 @@ object FlowSpec {
}
@nowarn // tests type assignments compile
-class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.receive=off\npekko.loglevel=INFO"))
{
+class FlowSpec extends StreamSpec(ConfigFactory.parseString("""
+ pekko.actor.debug.receive = off
+ pekko.loglevel = INFO
+ pekko.stream.materializer.initial-input-buffer-size = 2
+ pekko.stream.materializer.max-input-buffer-size = 2
+ """)) {
import FlowSpec._
- val settings = ActorMaterializerSettings(system).withInputBuffer(initialSize
= 2, maxSize = 2)
+ val settings = ActorMaterializerSettings(system)
implicit val materializer: Materializer = ActorMaterializer(settings)
val identity: Flow[Any, Any, NotUsed] => Flow[Any, Any, NotUsed] = in =>
in.map(e => e)
val identity2: Flow[Any, Any, NotUsed] => Flow[Any, Any, NotUsed] = in =>
identity(in)
+ private def withInputBuffer[In, Out, M](initialSize: Int, maxSize: Int)(
+ stream: Flow[In, In, NotUsed] => Flow[In, Out, M]): Flow[In, In,
NotUsed] => Flow[In, Out, M] =
+ flow => stream(flow).withAttributes(Attributes.inputBuffer(initialSize,
maxSize))
+
val toPublisher: (Source[Any, ?], Materializer) => Publisher[Any] =
(f, m) => f.runWith(Sink.asPublisher(false))(m)
@@ -72,8 +81,8 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
for ((name, op) <- List("identity" -> identity, "identity2" -> identity2);
n <- List(1, 2, 4)) {
s"request initial elements from upstream ($name, $n)" in {
- new ChainSetup(op, settings.withInputBuffer(initialSize = n, maxSize =
n), toPublisher) {
- upstream.expectRequest(upstreamSubscription,
this.settings.maxInputBufferSize)
+ new ChainSetup(withInputBuffer(n, n)(op), settings, toPublisher) {
+ upstream.expectRequest(upstreamSubscription, n)
}
}
}
@@ -123,7 +132,7 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
}
"cancel upstream when single subscriber cancels subscription while
receiving data" in {
- new ChainSetup(identity, settings.withInputBuffer(initialSize = 1,
maxSize = 1), toPublisher) {
+ new ChainSetup(withInputBuffer(1, 1)(identity), settings, toPublisher) {
downstreamSubscription.request(5)
upstreamSubscription.expectRequest(1)
upstreamSubscription.sendNext("test")
@@ -338,7 +347,7 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
"A Flow with multiple subscribers (FanOutBox)" must {
"adapt speed to the currently slowest subscriber" in {
- new ChainSetup(identity, settings.withInputBuffer(initialSize = 1,
maxSize = 1), toFanoutPublisher(1)) {
+ new ChainSetup(withInputBuffer(1, 1)(identity), settings,
toFanoutPublisher(1)) {
val downstream2 = TestSubscriber.manualProbe[Any]()
publisher.subscribe(downstream2)
val downstream2Subscription = downstream2.expectSubscription()
@@ -364,7 +373,7 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
}
"support slow subscriber with fan-out 2" in {
- new ChainSetup(identity, settings.withInputBuffer(initialSize = 1,
maxSize = 1), toFanoutPublisher(2)) {
+ new ChainSetup(withInputBuffer(1, 1)(identity), settings,
toFanoutPublisher(2)) {
val downstream2 = TestSubscriber.manualProbe[Any]()
publisher.subscribe(downstream2)
val downstream2Subscription = downstream2.expectSubscription()
@@ -403,7 +412,7 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
}
"support incoming subscriber while elements were requested before" in {
- new ChainSetup(identity, settings.withInputBuffer(initialSize = 1,
maxSize = 1), toFanoutPublisher(1)) {
+ new ChainSetup(withInputBuffer(1, 1)(identity), settings,
toFanoutPublisher(1)) {
downstreamSubscription.request(5)
upstream.expectRequest(upstreamSubscription, 1)
upstreamSubscription.sendNext("a1")
@@ -440,7 +449,7 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
}
"be unblocked when blocking subscriber cancels subscription" in {
- new ChainSetup(identity, settings.withInputBuffer(initialSize = 1,
maxSize = 1), toFanoutPublisher(1)) {
+ new ChainSetup(withInputBuffer(1, 1)(identity), settings,
toFanoutPublisher(1)) {
val downstream2 = TestSubscriber.manualProbe[Any]()
publisher.subscribe(downstream2)
val downstream2Subscription = downstream2.expectSubscription()
@@ -476,7 +485,7 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
}
"call future subscribers' onError after onSubscribe if initial upstream
was completed" in {
- new ChainSetup(identity, settings.withInputBuffer(initialSize = 1,
maxSize = 1), toFanoutPublisher(1)) {
+ new ChainSetup(withInputBuffer(1, 1)(identity), settings,
toFanoutPublisher(1)) {
val downstream2 = TestSubscriber.manualProbe[Any]()
// don't link it just yet
@@ -515,8 +524,8 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
"call future subscribers' onError should be called instead of onSubscribed
after initial upstream reported an error" in {
new ChainSetup[Int, String, NotUsed](
- _.map(_ => throw TestException),
- settings.withInputBuffer(initialSize = 1, maxSize = 1),
+ withInputBuffer(1, 1)(_.map(_ => throw TestException)),
+ settings,
toFanoutPublisher(1)) {
downstreamSubscription.request(1)
upstreamSubscription.expectRequest(1)
@@ -533,7 +542,7 @@ class FlowSpec extends
StreamSpec(ConfigFactory.parseString("pekko.actor.debug.r
}
"call future subscribers' onError when all subscriptions were cancelled"
in {
- new ChainSetup(identity, settings.withInputBuffer(initialSize = 1,
maxSize = 1), toFanoutPublisher(16)) {
+ new ChainSetup(withInputBuffer(1, 1)(identity), settings,
toFanoutPublisher(16)) {
upstreamSubscription.expectRequest(1)
downstreamSubscription.cancel()
upstreamSubscription.expectCancellation()
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithIndexSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithIndexSpec.scala
index 7b696020ea..a4a319a108 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithIndexSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithIndexSpec.scala
@@ -18,8 +18,14 @@ import pekko.stream.{ ActorMaterializer,
ActorMaterializerSettings, ClosedShape,
import pekko.stream.testkit.{ StreamSpec, TestSubscriber }
import scala.annotation.nowarn
+import com.typesafe.config.ConfigFactory
+
@nowarn // keep unused imports
-class FlowZipWithIndexSpec extends StreamSpec {
+class FlowZipWithIndexSpec
+ extends StreamSpec(ConfigFactory.parseString("""
+ pekko.stream.materializer.initial-input-buffer-size = 2
+ pekko.stream.materializer.max-input-buffer-size = 16
+ """)) {
//#zip-with-index
import org.apache.pekko
@@ -27,7 +33,7 @@ class FlowZipWithIndexSpec extends StreamSpec {
import pekko.stream.scaladsl.Sink
//#zip-with-index
- val settings = ActorMaterializerSettings(system).withInputBuffer(initialSize
= 2, maxSize = 16)
+ val settings = ActorMaterializerSettings(system)
implicit val materializer: Materializer = ActorMaterializer(settings)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphBackedFlowSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphBackedFlowSpec.scala
index 4ea1560961..df8328d1d9 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphBackedFlowSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/GraphBackedFlowSpec.scala
@@ -23,6 +23,8 @@ import pekko.stream.testkit._
import org.reactivestreams.Subscriber
+import com.typesafe.config.ConfigFactory
+
object GraphFlowSpec {
val source1 = Source(0 to 3)
@@ -50,11 +52,15 @@ object GraphFlowSpec {
}
@nowarn
-class GraphFlowSpec extends StreamSpec {
+class GraphFlowSpec
+ extends StreamSpec(ConfigFactory.parseString("""
+ pekko.stream.materializer.initial-input-buffer-size = 2
+ pekko.stream.materializer.max-input-buffer-size = 16
+ """)) {
import GraphFlowSpec._
- val settings = ActorMaterializerSettings(system).withInputBuffer(initialSize
= 2, maxSize = 16)
+ val settings = ActorMaterializerSettings(system)
implicit val materializer: Materializer = ActorMaterializer(settings)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/TakeLastSinkSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/TakeLastSinkSpec.scala
index 88da15d8b0..3103a5d34e 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/TakeLastSinkSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/TakeLastSinkSpec.scala
@@ -21,10 +21,16 @@ import org.apache.pekko
import pekko.stream.{ AbruptTerminationException, ActorMaterializer,
ActorMaterializerSettings, Materializer }
import pekko.stream.testkit.{ StreamSpec, TestPublisher }
+import com.typesafe.config.ConfigFactory
+
@nowarn
-class TakeLastSinkSpec extends StreamSpec {
+class TakeLastSinkSpec
+ extends StreamSpec(ConfigFactory.parseString("""
+ pekko.stream.materializer.initial-input-buffer-size = 2
+ pekko.stream.materializer.max-input-buffer-size = 16
+ """)) {
- val settings = ActorMaterializerSettings(system).withInputBuffer(initialSize
= 2, maxSize = 16)
+ val settings = ActorMaterializerSettings(system)
implicit val mat: Materializer = ActorMaterializer(settings)
diff --git
a/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-materializer-settings.excludes
b/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-materializer-settings.excludes
new file mode 100644
index 0000000000..b8885d0f79
--- /dev/null
+++
b/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-materializer-settings.excludes
@@ -0,0 +1,32 @@
+# 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.
+
+# Remove deprecated ActorMaterializerSettings builder methods and
materializer-specific settings classes
+# (deprecated since Akka 2.6.0)
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.ActorMaterializerSettings.withInputBuffer")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.ActorMaterializerSettings.withDispatcher")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.ActorMaterializerSettings.ioSettings")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.ActorMaterializerSettings.withIOSettings")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.ActorMaterializerSettings.streamRefSettings")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.ActorMaterializerSettings.withStreamRefSettings")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.IOSettings.this")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.IOSettings")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.IOSettings$")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.StreamRefSettings")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.StreamRefSettings$")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.impl.streamref.StreamRefSettingsImpl")
+ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.impl.streamref.StreamRefSettingsImpl$")
diff --git
a/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes
b/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes
index dc9a61a29e..c2545a8da5 100644
---
a/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes
+++
b/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes
@@ -38,11 +38,7 @@
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.Bind
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.BindFailedException.productArity")
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.BindFailedException.productPrefix")
ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.FanInShape1N")
-ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.IOSettings.create")
-ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.IOSettings.apply")
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.Materializer.schedulePeriodically")
-ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.StreamRefSettings.create")
-ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.StreamRefSettings.apply")
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.UniformFanInShape.inSeq")
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.UniformFanOutShape.outArray")
ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.impl.ConstantFun*")
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/ActorMaterializer.scala
b/stream/src/main/scala/org/apache/pekko/stream/ActorMaterializer.scala
index 3e234c5069..eb14650226 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/ActorMaterializer.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/ActorMaterializer.scala
@@ -26,6 +26,7 @@ import pekko.actor.ActorSystem
import pekko.actor.ExtendedActorSystem
import pekko.annotation.InternalApi
import pekko.stream.impl._
+import pekko.stream.impl.streamref.StreamRefDefaultSettings
import pekko.stream.stage.GraphStageLogic
import pekko.util.Helpers.toRootLowerCase
@@ -196,7 +197,10 @@ private[pekko] object ActorMaterializerSettings {
@deprecated(
"Use config or attributes to configure the materializer. See migration
guide for details
https://doc.akka.io/docs/akka/2.6/project/migration-guide-2.5.x-2.6.x.html";,
"Akka 2.6.0")
- def apply(config: Config): ActorMaterializerSettings =
+ def apply(config: Config): ActorMaterializerSettings = {
+ val ioConfig = config.getConfig("io")
+ val streamRefConfig = config.getConfig("stream-ref")
+
new ActorMaterializerSettings(
initialInputBufferSize = config.getInt("initial-input-buffer-size"),
maxInputBufferSize = config.getInt("max-input-buffer-size"),
@@ -209,9 +213,17 @@ private[pekko] object ActorMaterializerSettings {
autoFusing = config.getBoolean("auto-fusing"),
maxFixedBufferSize = config.getInt("max-fixed-buffer-size"),
syncProcessingLimit = config.getInt("sync-processing-limit"),
- ioSettings = IOSettings(config.getConfig("io")),
- streamRefSettings = StreamRefSettings(config.getConfig("stream-ref")),
+ tcpWriteBufferSize = math.min(Int.MaxValue,
ioConfig.getBytes("tcp.write-buffer-size")).toInt,
+ coalesceWrites = ioConfig.getInt("tcp.coalesce-writes"),
+ streamRefDefaultSettings = StreamRefDefaultSettings(
+ bufferCapacity = streamRefConfig.getInt("buffer-capacity"),
+ demandRedeliveryInterval =
+ streamRefConfig.getDuration("demand-redelivery-interval",
TimeUnit.MILLISECONDS).millis,
+ subscriptionTimeout =
streamRefConfig.getDuration("subscription-timeout",
TimeUnit.MILLISECONDS).millis,
+ finalTerminationSignalDeadline =
+ streamRefConfig.getDuration("final-termination-signal-deadline",
TimeUnit.MILLISECONDS).millis),
blockingIoDispatcher = config.getString("blocking-io-dispatcher"))
+ }
}
/**
@@ -238,8 +250,9 @@ final class ActorMaterializerSettings @InternalApi private (
private[stream] val autoFusing: Boolean,
private[stream] val maxFixedBufferSize: Int,
private[stream] val syncProcessingLimit: Int,
- val ioSettings: IOSettings,
- val streamRefSettings: StreamRefSettings,
+ private[stream] val tcpWriteBufferSize: Int,
+ private[stream] val coalesceWrites: Int,
+ private[stream] val streamRefDefaultSettings: StreamRefDefaultSettings,
private[stream] val blockingIoDispatcher: String) {
require(initialInputBufferSize > 0, "initialInputBufferSize must be > 0")
@@ -262,8 +275,9 @@ final class ActorMaterializerSettings @InternalApi private (
autoFusing: Boolean = this.autoFusing,
maxFixedBufferSize: Int = this.maxFixedBufferSize,
syncProcessingLimit: Int = this.syncProcessingLimit,
- ioSettings: IOSettings = this.ioSettings,
- streamRefSettings: StreamRefSettings = this.streamRefSettings,
+ tcpWriteBufferSize: Int = this.tcpWriteBufferSize,
+ coalesceWrites: Int = this.coalesceWrites,
+ streamRefDefaultSettings: StreamRefDefaultSettings =
this.streamRefDefaultSettings,
blockingIoDispatcher: String = this.blockingIoDispatcher) = {
new ActorMaterializerSettings(
initialInputBufferSize,
@@ -277,37 +291,15 @@ final class ActorMaterializerSettings @InternalApi
private (
autoFusing,
maxFixedBufferSize,
syncProcessingLimit,
- ioSettings,
- streamRefSettings,
+ tcpWriteBufferSize,
+ coalesceWrites,
+ streamRefDefaultSettings,
blockingIoDispatcher)
}
- /**
- * Each asynchronous piece of a materialized stream topology is executed by
one Actor
- * that manages an input buffer for all inlets of its shape. This setting
configures
- * the default for initial and maximal input buffer in number of elements
for each inlet.
- * This can be overridden for individual parts of the
- * stream topology by using [[pekko.stream.Attributes#inputBuffer]].
- *
- * FIXME: this is used for all kinds of buffers, not only the stream actor,
some use initial some use max,
- * document and or fix if it should not be like that. Search for
get[Attributes.InputBuffer] to see how it is used
- */
- @deprecated("Use attribute 'Attributes.InputBuffer' to change setting
value", "Akka 2.6.0")
- def withInputBuffer(initialSize: Int, maxSize: Int):
ActorMaterializerSettings = {
- if (initialSize == this.initialInputBufferSize && maxSize ==
this.maxInputBufferSize) this
- else copy(initialInputBufferSize = initialSize, maxInputBufferSize =
maxSize)
- }
-
- /**
- * This setting configures the default dispatcher to be used by streams
materialized
- * with the [[ActorMaterializer]]. This can be overridden for individual
parts of the
- * stream topology by using [[pekko.stream.Attributes#dispatcher]].
- */
- @deprecated("Use attribute 'ActorAttributes.Dispatcher' to change setting
value", "Akka 2.6.0")
- def withDispatcher(dispatcher: String): ActorMaterializerSettings = {
+ private[pekko] def withDispatcher(dispatcher: String):
ActorMaterializerSettings =
if (this.dispatcher == dispatcher) this
else copy(dispatcher = dispatcher)
- }
/**
* Leaked publishers and subscribers are cleaned up when they are not used
within a given
@@ -317,15 +309,6 @@ final class ActorMaterializerSettings @InternalApi private
(
if (settings == this.subscriptionTimeoutSettings) this
else copy(subscriptionTimeoutSettings = settings)
- def withIOSettings(ioSettings: IOSettings): ActorMaterializerSettings =
- if (ioSettings == this.ioSettings) this
- else copy(ioSettings = ioSettings)
-
- /** Change settings specific to [[SourceRef]] and [[SinkRef]]. */
- def withStreamRefSettings(streamRefSettings: StreamRefSettings):
ActorMaterializerSettings =
- if (streamRefSettings == this.streamRefSettings) this
- else copy(streamRefSettings = streamRefSettings)
-
private def requirePowerOfTwo(n: Integer, name: String): Unit = {
require(n > 0, s"$name must be > 0")
require((n & (n - 1)) == 0, s"$name must be a power of two")
@@ -343,7 +326,8 @@ final class ActorMaterializerSettings @InternalApi private (
s.syncProcessingLimit == syncProcessingLimit &&
s.fuzzingMode == fuzzingMode &&
s.autoFusing == autoFusing &&
- s.ioSettings == ioSettings &&
+ s.tcpWriteBufferSize == tcpWriteBufferSize &&
+ s.coalesceWrites == coalesceWrites &&
s.blockingIoDispatcher == blockingIoDispatcher
case _ => false
}
@@ -373,45 +357,7 @@ final class ActorMaterializerSettings @InternalApi private
(
override def toString: String =
s"ActorMaterializerSettings($initialInputBufferSize,$maxInputBufferSize," +
s"$dispatcher,$supervisionDecider,$subscriptionTimeoutSettings,$debugLogging,$outputBurstLimit,"
+
- s"$syncProcessingLimit,$fuzzingMode,$autoFusing,$ioSettings)"
-}
-
-@InternalApi
-private[pekko] object IOSettings {
- @deprecated(
- "Use setting 'pekko.stream.materializer.io.tcp.write-buffer-size' or
attribute TcpAttributes.writeBufferSize instead",
- "Akka 2.6.0")
- def apply(config: Config): IOSettings =
- new IOSettings(
- tcpWriteBufferSize = math.min(Int.MaxValue,
config.getBytes("tcp.write-buffer-size")).toInt,
- coalesceWrites = config.getInt("tcp.coalesce-writes"))
-}
-
-final class IOSettings private (
- private[stream] val tcpWriteBufferSize: Int,
- val coalesceWrites: Int) {
-
- // constructor for binary compatibility with version 2.6.15 and earlier
- @deprecated("Use attribute 'TcpAttributes.TcpWriteBufferSize' to read the
concrete setting value", "Akka 2.6.0")
- def this(tcpWriteBufferSize: Int) = this(tcpWriteBufferSize, coalesceWrites
= 10)
-
- def withTcpWriteBufferSize(value: Int): IOSettings = copy(tcpWriteBufferSize
= value)
-
- def withCoalesceWrites(value: Int): IOSettings = copy(coalesceWrites = value)
-
- private def copy(tcpWriteBufferSize: Int = tcpWriteBufferSize,
coalesceWrites: Int = coalesceWrites): IOSettings =
- new IOSettings(tcpWriteBufferSize, coalesceWrites)
-
- override def equals(other: Any): Boolean = other match {
- case s: IOSettings => s.tcpWriteBufferSize == tcpWriteBufferSize &&
s.coalesceWrites == coalesceWrites
- case _ => false
- }
-
- override def hashCode(): Int =
- 31 * tcpWriteBufferSize + coalesceWrites
-
- override def toString =
- s"""IoSettings($tcpWriteBufferSize,$coalesceWrites)"""
+
s"$syncProcessingLimit,$fuzzingMode,$autoFusing,$tcpWriteBufferSize,$coalesceWrites)"
}
object StreamSubscriptionTimeoutSettings {
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/StreamRefSettings.scala
b/stream/src/main/scala/org/apache/pekko/stream/StreamRefSettings.scala
deleted file mode 100644
index 6aa90497ae..0000000000
--- a/stream/src/main/scala/org/apache/pekko/stream/StreamRefSettings.scala
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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) 2018-2022 Lightbend Inc. <https://www.lightbend.com>
- */
-
-package org.apache.pekko.stream
-
-import java.util.concurrent.TimeUnit
-
-import scala.annotation.nowarn
-import scala.concurrent.duration._
-
-import org.apache.pekko
-import pekko.annotation.{ DoNotInherit, InternalApi }
-import pekko.stream.impl.streamref.StreamRefSettingsImpl
-
-import com.typesafe.config.Config
-
-@InternalApi
-private[stream] object StreamRefSettings {
-
- /** Scala API */
- @deprecated(
- "Use attributes on the Runnable graph or change the defaults in
configuration, see migration guide for details
https://doc.akka.io/docs/akka/2.6/project/migration-guide-2.5.x-2.6.x.html";,
- since = "Akka 2.6.0")
- def apply(c: Config): StreamRefSettings = {
- StreamRefSettingsImpl(
- bufferCapacity = c.getInt("buffer-capacity"),
- demandRedeliveryInterval = c.getDuration("demand-redelivery-interval",
TimeUnit.MILLISECONDS).millis,
- subscriptionTimeout = c.getDuration("subscription-timeout",
TimeUnit.MILLISECONDS).millis,
- finalTerminationSignalDeadline =
c.getDuration("final-termination-signal-deadline",
TimeUnit.MILLISECONDS).millis)
- }
-}
-
-/**
- * Settings specific to [[SourceRef]] and [[SinkRef]].
- * More detailed documentation about each of the settings is available in
`reference.conf`.
- */
-@DoNotInherit
-@nowarn("msg=deprecated")
-trait StreamRefSettings {
- @deprecated("Use attribute 'StreamRefAttributes.BufferCapacity' to read the
concrete setting value", "Akka 2.6.0")
- def bufferCapacity: Int
- @deprecated(
- "Use attribute 'StreamRefAttributes.DemandRedeliveryInterval' to read the
concrete setting value",
- "Akka 2.6.0")
- def demandRedeliveryInterval: FiniteDuration
- @deprecated("Use attribute 'StreamRefAttributes.SubscriptionTimeout' to read
the concrete setting value",
- "Akka 2.6.0")
- def subscriptionTimeout: FiniteDuration
- @deprecated(
- "Use attribute 'StreamRefAttributes.FinalTerminationSignalDeadline' to
read the concrete setting value",
- "Akka 2.6.0")
- def finalTerminationSignalDeadline: FiniteDuration
-
- // --- with... methods ---
-
- def withBufferCapacity(value: Int): StreamRefSettings
- def withDemandRedeliveryInterval(value: FiniteDuration): StreamRefSettings
- def withSubscriptionTimeout(value: FiniteDuration): StreamRefSettings
- def withTerminationReceivedBeforeCompletionLeeway(value: FiniteDuration):
StreamRefSettings
-}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala
index 79c95e9383..75f855b3a2 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/TcpStages.scala
@@ -268,8 +268,8 @@ private[stream] object ConnectionSourceStage {
@nowarn("msg=deprecated")
private val writeBufferSize = inheritedAttributes
- .get[TcpAttributes.TcpWriteBufferSize](
-
TcpAttributes.TcpWriteBufferSize(eagerMaterializer.settings.ioSettings.tcpWriteBufferSize))
+ .get[TcpAttributes.TcpWriteBufferSize](TcpAttributes.TcpWriteBufferSize(
+ eagerMaterializer.settings.tcpWriteBufferSize))
.size
private var writeBuffer = ByteString.empty
@@ -280,7 +280,7 @@ private[stream] object ConnectionSourceStage {
private var connectionClosePending = false
@nowarn("msg=deprecated")
- private val coalesceWrites =
eagerMaterializer.settings.ioSettings.coalesceWrites
+ private val coalesceWrites = eagerMaterializer.settings.coalesceWrites
private def coalesceWritesDisabled = coalesceWrites == 0
private var writeDelayCountDown = 0
private var previousWriteBufferSize = 0
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SinkRefImpl.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SinkRefImpl.scala
index 614a1f8f64..2be9ac1b28 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SinkRefImpl.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SinkRefImpl.scala
@@ -13,6 +13,8 @@
package org.apache.pekko.stream.impl.streamref
+import java.util.concurrent.atomic.AtomicBoolean
+
import scala.annotation.nowarn
import scala.util.{ Failure, Success, Try }
@@ -27,8 +29,6 @@ import pekko.stream.scaladsl.Sink
import pekko.stream.stage._
import pekko.util.{ OptionVal, PrettyDuration }
-import java.util.concurrent.atomic.AtomicBoolean
-
/** INTERNAL API: Implementation class, not intended to be touched directly by
end-users */
@InternalApi
private[stream] final case class SinkRefImpl[In](initialPartnerRef: ActorRef)
extends SinkRef[In] {
@@ -88,18 +88,18 @@ private[stream] final class SinkRefStageImpl[In]
private[pekko] (
private val streamRefsMaster = StreamRefsMaster(eagerMaterializer.system)
// settings ---
- @nowarn("msg=deprecated") // can't remove this settings access without
breaking compat
private val subscriptionTimeout = {
import StreamRefAttributes._
- val settings = eagerMaterializer.settings.streamRefSettings
+ @nowarn("cat=deprecation") // Materializer.settings is needed for
stream-ref default settings.
+ val settings = eagerMaterializer.settings.streamRefDefaultSettings
inheritedAttributes.get[StreamRefAttributes.SubscriptionTimeout](
SubscriptionTimeout(settings.subscriptionTimeout))
}
- @nowarn("msg=deprecated") // can't remove this settings access without
breaking compat
private val finalTerminationSignalDeadline = {
import StreamRefAttributes._
- val settings = eagerMaterializer.settings.streamRefSettings
+ @nowarn("cat=deprecation") // Materializer.settings is needed for
stream-ref default settings.
+ val settings = eagerMaterializer.settings.streamRefDefaultSettings
inheritedAttributes
.get[StreamRefAttributes.FinalTerminationSignalDeadline](
FinalTerminationSignalDeadline(settings.finalTerminationSignalDeadline))
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SourceRefImpl.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SourceRefImpl.scala
index 59d383c5ad..284ec19a28 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SourceRefImpl.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/SourceRefImpl.scala
@@ -13,6 +13,8 @@
package org.apache.pekko.stream.impl.streamref
+import java.util.concurrent.atomic.AtomicBoolean
+
import scala.annotation.nowarn
import org.apache.pekko
@@ -26,8 +28,6 @@ import pekko.stream.scaladsl.Source
import pekko.stream.stage._
import pekko.util.{ OptionVal, PrettyDuration }
-import java.util.concurrent.atomic.AtomicBoolean
-
/** INTERNAL API: Implementation class, not intended to be touched directly by
end-users */
@InternalApi
private[stream] final case class SourceRefImpl[T](initialPartnerRef: ActorRef)
extends SourceRef[T] {
@@ -148,24 +148,20 @@ private[stream] final class SourceRefStageImpl[Out](
// settings ---
import StreamRefAttributes._
- @nowarn("msg=deprecated") // can't remove this settings access without
breaking compat
- private val settings = eagerMaterializer.settings.streamRefSettings
+ @nowarn("cat=deprecation") // Materializer.settings is needed for
stream-ref default settings.
+ private val settings =
eagerMaterializer.settings.streamRefDefaultSettings
- @nowarn("msg=deprecated") // can't remove this settings access without
breaking compat
private val subscriptionTimeout =
inheritedAttributes.get[StreamRefAttributes.SubscriptionTimeout](
SubscriptionTimeout(settings.subscriptionTimeout))
- @nowarn("msg=deprecated") // can't remove this settings access without
breaking compat
private val bufferCapacity = inheritedAttributes
.get[StreamRefAttributes.BufferCapacity](StreamRefAttributes.BufferCapacity(settings.bufferCapacity))
.capacity
- @nowarn("msg=deprecated") // can't remove this settings access without
breaking compat
private val demandRedeliveryInterval = inheritedAttributes
.get[StreamRefAttributes.DemandRedeliveryInterval](DemandRedeliveryInterval(settings.demandRedeliveryInterval))
.timeout
- @nowarn("msg=deprecated") // can't remove this settings access without
breaking compat
private val finalTerminationSignalDeadline =
inheritedAttributes
.get[StreamRefAttributes.FinalTerminationSignalDeadline](
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefDefaultSettings.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefDefaultSettings.scala
new file mode 100644
index 0000000000..d16ab30290
--- /dev/null
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefDefaultSettings.scala
@@ -0,0 +1,27 @@
+/*
+ * 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) 2018-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.stream.impl.streamref
+
+import scala.concurrent.duration.FiniteDuration
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+
+/** INTERNAL API */
+@InternalApi
+private[pekko] final case class StreamRefDefaultSettings(
+ bufferCapacity: Int,
+ demandRedeliveryInterval: FiniteDuration,
+ subscriptionTimeout: FiniteDuration,
+ finalTerminationSignalDeadline: FiniteDuration)
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefSettingsImpl.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefSettingsImpl.scala
deleted file mode 100644
index 7701e386b0..0000000000
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/streamref/StreamRefSettingsImpl.scala
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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) 2018-2022 Lightbend Inc. <https://www.lightbend.com>
- */
-
-package org.apache.pekko.stream.impl.streamref
-
-import scala.concurrent.duration.FiniteDuration
-
-import org.apache.pekko
-import pekko.annotation.InternalApi
-import pekko.event.Logging
-import pekko.stream.StreamRefSettings
-
-/** INTERNAL API */
-@InternalApi
-private[pekko] final case class StreamRefSettingsImpl(
- override val bufferCapacity: Int,
- override val demandRedeliveryInterval: FiniteDuration,
- override val subscriptionTimeout: FiniteDuration,
- override val finalTerminationSignalDeadline: FiniteDuration)
- extends StreamRefSettings {
-
- override def withBufferCapacity(value: Int): StreamRefSettings =
copy(bufferCapacity = value)
- override def withDemandRedeliveryInterval(value: FiniteDuration):
StreamRefSettings =
- copy(demandRedeliveryInterval = value)
- override def withSubscriptionTimeout(value: FiniteDuration):
StreamRefSettings = copy(subscriptionTimeout = value)
- override def withTerminationReceivedBeforeCompletionLeeway(value:
FiniteDuration): StreamRefSettings =
- copy(finalTerminationSignalDeadline = value)
-
- override def productPrefix: String =
Logging.simpleName(classOf[StreamRefSettings])
-
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
