This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch remove-actor-materializer-deprecated in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 7fd5eda910dd3efbd8a5720a677c72075bb68463 Author: 虎鸣 <[email protected]> AuthorDate: Sun Jul 5 17:01:00 2026 +0800 refactor: remove deprecated ActorMaterializerSettings, IOSettings and StreamRefSettings factory methods Motivation: Several ActorMaterializerSettings builder methods (withInputBuffer, withDispatcher), IOSettings factory/constructor, and StreamRefSettings factory were deprecated since Akka 2.6.0. Since Pekko 2.0.0 is a major version bump, these can be safely removed. Modification: - Remove ActorMaterializerSettings.withInputBuffer (deprecated since Akka 2.6.0) - Change ActorMaterializerSettings.withDispatcher from public deprecated to private[pekko] (still used internally by remote module ArterySettings) - Remove IOSettings.apply(config) factory and deprecated binary-compat constructor - Change IOSettings constructor from private to private[stream] for internal use - Remove StreamRefSettings.apply(config) factory - Inline IOSettings and StreamRefSettings construction in ActorMaterializerSettings.apply - Add MiMa exclusion filters for the removed methods Result: Deprecated Akka 2.6.0 era APIs are removed from the public surface while preserving internal functionality used by the remote module. Tests: - sbt "stream / Compile / compile" "remote / Compile / compile" - success - sbt "stream / Test / compile" "remote / Test / compile" - success - sbt "+stream / mimaReportBinaryIssues" "+remote / mimaReportBinaryIssues" - no issues References: None - proactive cleanup of deprecated APIs before Pekko 2.0.0 release --- ...emove-deprecated-materializer-settings.excludes | 21 ++++++++ .../apache/pekko/stream/ActorMaterializer.scala | 59 +++++++--------------- .../apache/pekko/stream/StreamRefSettings.scala | 20 +------- 3 files changed, 41 insertions(+), 59 deletions(-) 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..e99832e516 --- /dev/null +++ b/stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-materializer-settings.excludes @@ -0,0 +1,21 @@ +# 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 IOSettings constructor (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.IOSettings.this") 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..0fb1878075 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.StreamRefSettingsImpl import pekko.stream.stage.GraphStageLogic import pekko.util.Helpers.toRootLowerCase @@ -209,8 +210,21 @@ 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")), + ioSettings = new IOSettings( + tcpWriteBufferSize = math.min(Int.MaxValue, config.getConfig("io").getBytes("tcp.write-buffer-size")).toInt, + coalesceWrites = config.getConfig("io").getInt("tcp.coalesce-writes")), + streamRefSettings = StreamRefSettingsImpl( + bufferCapacity = config.getConfig("stream-ref").getInt("buffer-capacity"), + demandRedeliveryInterval = config + .getConfig("stream-ref") + .getDuration("demand-redelivery-interval", TimeUnit.MILLISECONDS) + .millis, + subscriptionTimeout = + config.getConfig("stream-ref").getDuration("subscription-timeout", TimeUnit.MILLISECONDS).millis, + finalTerminationSignalDeadline = config + .getConfig("stream-ref") + .getDuration("final-termination-signal-deadline", TimeUnit.MILLISECONDS) + .millis), blockingIoDispatcher = config.getString("blocking-io-dispatcher")) } @@ -282,32 +296,9 @@ final class ActorMaterializerSettings @InternalApi private ( 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 @@ -377,24 +368,12 @@ final class ActorMaterializerSettings @InternalApi private ( } @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")) -} +private[pekko] object IOSettings -final class IOSettings private ( +final class IOSettings private[stream] ( 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) diff --git a/stream/src/main/scala/org/apache/pekko/stream/StreamRefSettings.scala b/stream/src/main/scala/org/apache/pekko/stream/StreamRefSettings.scala index 6aa90497ae..da2ad89c7b 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/StreamRefSettings.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/StreamRefSettings.scala @@ -13,32 +13,14 @@ 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) - } -} +private[stream] object StreamRefSettings {} /** * Settings specific to [[SourceRef]] and [[SinkRef]]. --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
