This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch pr-2878-clean in repository https://gitbox.apache.org/repos/asf/pekko.git
commit cc73eced6608d94b035aa9a9b018b1a620c6ab8c Author: He-Pin <[email protected]> AuthorDate: Sat Apr 25 18:28:08 2026 +0800 feat: wire TlsGraphStage routing layer + JVM-global config switch Motivation: PR #2878 (issue #2860) introduced TlsGraphStage as an internal pure GraphStage replacement for the legacy actor-based TLS path. Until this commit, TlsGraphStage was reachable only by directly instantiating it; TLS.apply() unconditionally returned the legacy TlsModule path. Without a routing seam, neither integration tests nor downstream callers could exercise the GraphStage implementation through the public API, and TLS 1.3 remained broken in the only path that was actually wired. Modification: - stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala: Add a `UseLegacyActor` boolean read once at class-init from system property `pekko.stream.materializer.tls.use-legacy-actor` (sysprop wins over config so tests can flip without rebuilding reference.conf). Route TLS.apply() through TlsModule (legacy) when true, or through TlsGraphStage wrapped in `Attributes.asyncBoundary` when false. - stream/src/main/resources/reference.conf: Document and default the new key under `pekko.stream.materializer.tls.use-legacy-actor = true` (legacy path remains the default in this PR; the GraphStage path is opt-in). Result: - Public TLS BidiFlow is now selectable between the legacy actor-backed implementation and the GraphStage implementation via a single JVM-wide switch, with no API or binary surface change. - Legacy path remains the default, preserving existing behavior for all current users until the GraphStage path passes the full TLS suite. - Enables empirical validation under `-Dpekko.stream.materializer.tls.use-legacy-actor=false`. References: - Issue #2860 - PR #2878 --- stream/src/main/resources/reference.conf | 15 ++++++++++ .../org/apache/pekko/stream/scaladsl/TLS.scala | 33 ++++++++++++++++++++-- 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/stream/src/main/resources/reference.conf b/stream/src/main/resources/reference.conf index bba71993ec..f22aded73c 100644 --- a/stream/src/main/resources/reference.conf +++ b/stream/src/main/resources/reference.conf @@ -140,6 +140,21 @@ pekko { # Time to wait for async materializer creation before throwing an exception creation-timeout = 20 seconds + # TLS implementation selector. + # + # When `true` (the current default) Pekko Streams routes TLS through the + # legacy actor-based path that depends on internal FanoutProcessor + + # ResizableMultiReaderRingBuffer infrastructure (slated for removal in 2.0). + # + # When `false` Pekko Streams routes TLS through TlsGraphStage, a pure + # GraphStage implementation that handles TLSv1.3 correctly and is faster + # for small/medium payloads (typical HTTP/gRPC). + # + # This switch is read once at class-init time of `org.apache.pekko.stream.scaladsl.TLS` + # and is JVM-global. Do not toggle it per-ActorSystem; the value reached by the + # first classload wins for the lifetime of the JVM. + tls.use-legacy-actor = true + //#stream-ref # configure defaults for SourceRef and SinkRef stream-ref { diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala index 88c2585d3d..7cb4967dcf 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/TLS.scala @@ -17,11 +17,13 @@ import javax.net.ssl.{ SSLContext, SSLEngine, SSLSession } import scala.util.{ Success, Try } +import com.typesafe.config.ConfigFactory + import org.apache.pekko import pekko.NotUsed import pekko.stream._ import pekko.stream.TLSProtocol._ -import pekko.stream.impl.io.TlsModule +import pekko.stream.impl.io.{ TlsGraphStage, TlsModule } import pekko.util.ByteString /** @@ -61,6 +63,26 @@ import pekko.util.ByteString */ object TLS { + /** + * INTERNAL API. + * + * Whether to route TLS through the legacy actor-based path (`TlsModule` → `TLSActor`) + * or the [[TlsGraphStage]] GraphStage path. Read once at class-init; defaults to `true` + * (legacy path). Can be overridden via system property of the same name or via the + * `reference.conf` / `application.conf` key `pekko.stream.materializer.tls.use-legacy-actor`. + * + * System property wins over configuration to allow tests to flip the switch without + * rebuilding config. + */ + private[pekko] val UseLegacyActor: Boolean = { + val key = "pekko.stream.materializer.tls.use-legacy-actor" + val sysProp = System.getProperty(key) + if (sysProp != null) sysProp.toBoolean + else + try ConfigFactory.load().getBoolean(key) + catch { case _: Throwable => true } + } + /** * Create a StreamTls [[pekko.stream.scaladsl.BidiFlow]]. * @@ -76,8 +98,13 @@ object TLS { createSSLEngine: () => SSLEngine, verifySession: SSLSession => Try[Unit], closing: TLSClosing): scaladsl.BidiFlow[SslTlsOutbound, ByteString, ByteString, SslTlsInbound, NotUsed] = - scaladsl.BidiFlow.fromGraph( - TlsModule(Attributes.none, () => createSSLEngine(), session => verifySession(session), closing)) + if (UseLegacyActor) + scaladsl.BidiFlow.fromGraph( + TlsModule(Attributes.none, () => createSSLEngine(), session => verifySession(session), closing)) + else + scaladsl.BidiFlow + .fromGraph(new TlsGraphStage(createSSLEngine, verifySession, closing)) + .withAttributes(Attributes.asyncBoundary) /** * Create a StreamTls [[pekko.stream.scaladsl.BidiFlow]]. --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
