refactor: remove deprecated ActorMaterializerSettings, IOSettings and StreamRefSettings factory methods

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)
 

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)
 

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)
 

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()

stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowZipWithIndexSpec.scala

@@ -18,16 +18,22 @@ 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
   import pekko.stream.scaladsl.Source
   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)
 

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)
 

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)
 

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$")

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*")