From 821058903bc2ee7feb71824451db449be9b65159 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Sun, 30 Aug 2026 12:45:02 +0100 Subject: [PATCH 1/3] http/2: prototype ALPN fallback from HTTP/2 to HTTP/1.1 for client connections The HTTP/2 client offers only `h2` over ALPN and then unconditionally builds the HTTP/2 stack, so a server without HTTP/2 support fails the stream. Adds `OutgoingConnectionBuilder.http2WithFallback()`, which offers both `h2` and `http/1.1` and installs the client layer matching what the server selected. The decision needs a "handshake complete" signal. ProtocolSwitch decides on the first inbound SessionBytes, which works server-side because the client always speaks first, but deadlocks here: on HTTP/1.1 the server stays silent until it gets a request and the switch would be holding that request. Neither the TLS stage nor SSLEngine offers such an event, so AlpnObservingSSLEngine watches getApplicationProtocol around wrap/unwrap and completes a promise that ClientProtocolSwitch waits on through an AsyncCallback. The stack is built inside Flow.fromMaterializer so each materialization gets its own engine and promise - PersistentConnection re-materializes the connection flow on every reconnect, so the closed-over var that httpsWithAlpn uses server-side would not work. Also fixes Http2JDKAlpnSupport.clientSetApplicationProtocols ignoring its `protocols` parameter and hardcoding Array("h2"), which is the seam this needs. Refs #1249, #483 Co-Authored-By: Claude Opus 5 (1M context) --- docs/src/main/paradox/client-side/http2.md | 28 ++- .../java/docs/http/javadsl/Http2Test.java | 4 + .../scala/docs/http/scaladsl/Http2Spec.scala | 3 + .../http2-client-alpn-fallback.excludes | 21 ++ .../engine/http2/AlpnObservingSSLEngine.scala | 119 +++++++++++ .../engine/http2/ClientProtocolSwitch.scala | 193 ++++++++++++++++++ .../pekko/http/impl/engine/http2/Http2.scala | 45 +++- .../impl/engine/http2/Http2AlpnSupport.scala | 2 +- .../http2/OutgoingConnectionBuilderImpl.scala | 12 ++ .../javadsl/OutgoingConnectionBuilder.scala | 12 ++ .../scaladsl/OutgoingConnectionBuilder.scala | 12 ++ .../http2/Http2ClientFallbackSpec.scala | 112 ++++++++++ 12 files changed, 558 insertions(+), 5 deletions(-) create mode 100644 http-core/src/main/mima-filters/2.0.x.backwards.excludes/http2-client-alpn-fallback.excludes create mode 100644 http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/AlpnObservingSSLEngine.scala create mode 100644 http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/ClientProtocolSwitch.scala create mode 100644 http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientFallbackSpec.scala diff --git a/docs/src/main/paradox/client-side/http2.md b/docs/src/main/paradox/client-side/http2.md index 59490bca0b..78ac6f11b7 100644 --- a/docs/src/main/paradox/client-side/http2.md +++ b/docs/src/main/paradox/client-side/http2.md @@ -14,9 +14,10 @@ unexpected when coming from a background with non-"streaming first" HTTP Clients ## Create the client -There are three mechanisms for a client to establish an HTTP/2 connection. Apache Pekko HTTP supports: +There are several mechanisms for a client to establish an HTTP/2 connection. Apache Pekko HTTP supports: - HTTP/2 over TLS + - HTTP/2 over TLS with ALPN fallback to HTTP/1.1 - HTTP/2 over a plain TCP connection ("h2c with prior knowledge") Apache Pekko HTTP doesn't support: @@ -36,7 +37,30 @@ Java HTTP/2 over TLS needs [Application-Layer Protocol Negotiation (ALPN)](https://en.wikipedia.org/wiki/Application-Layer_Protocol_Negotiation) to negotiate whether both client and server support HTTP/2. -Apache Pekko HTTP does not currently support protocol negotiation to fall back to HTTP/1.1 for this API. When the server does not support HTTP/2, the stream will fail. +`http2()` offers only `h2` in that handshake, so when the server does not support HTTP/2 the stream will fail. Use +`http2WithFallback()` if you need the connection to survive that case. + +### HTTP/2 over TLS with fallback to HTTP/1.1 + +@@@ warning +`http2WithFallback()` is available as a preview. This means it is ready to be evaluated, but the API and behavior +are likely to change. +@@@ + +`http2WithFallback()` offers both `h2` and `http/1.1` in the ALPN handshake and runs whichever protocol the server +selected. A server that speaks HTTP/2 gets an HTTP/2 connection; a server that does not - including one that ignores +ALPN entirely - gets an HTTP/1.1 connection instead of a failed stream: + +Scala +: @@snip[Http2Spec.scala](/docs/src/test/scala/docs/http/scaladsl/Http2Spec.scala) { #http2ClientWithFallback } + +Java +: @@snip[Http2Test.java](/docs/src/test/java/docs/http/javadsl/Http2Test.java) { #http2ClientWithFallback } + +Because the protocol is only known once the connection is up, requests should carry a @apidoc[RequestResponseAssociation] +as described in @ref[Request-response ordering](#request-response-ordering) - the flow may end up running HTTP/2, where +responses are not guaranteed to arrive in request order. + ### h2c with prior knowledge The other option is to connect and start communicating in HTTP/2 immediately. You must know beforehand the target server diff --git a/docs/src/test/java/docs/http/javadsl/Http2Test.java b/docs/src/test/java/docs/http/javadsl/Http2Test.java index 0dbd941c73..65ea9461f1 100644 --- a/docs/src/test/java/docs/http/javadsl/Http2Test.java +++ b/docs/src/test/java/docs/http/javadsl/Http2Test.java @@ -61,6 +61,10 @@ void testBindAndHandleAsync() { Http.get(system).connectionTo("127.0.0.1").toPort(8443).http2(); // #http2Client + // #http2ClientWithFallback + Http.get(system).connectionTo("127.0.0.1").toPort(8443).http2WithFallback(); + // #http2ClientWithFallback + // #http2ClientWithPriorKnowledge Http.get(system).connectionTo("127.0.0.1").toPort(8080).http2WithPriorKnowledge(); // #http2ClientWithPriorKnowledge diff --git a/docs/src/test/scala/docs/http/scaladsl/Http2Spec.scala b/docs/src/test/scala/docs/http/scaladsl/Http2Spec.scala index 1de73295d7..0f9d212db9 100644 --- a/docs/src/test/scala/docs/http/scaladsl/Http2Spec.scala +++ b/docs/src/test/scala/docs/http/scaladsl/Http2Spec.scala @@ -75,6 +75,9 @@ object Http2Spec { // #http2Client Http().connectionTo("localhost").toPort(8443).http2() // #http2Client + // #http2ClientWithFallback + Http().connectionTo("localhost").toPort(8443).http2WithFallback() + // #http2ClientWithFallback // #http2ClientWithPriorKnowledge Http().connectionTo("localhost").toPort(8080).http2WithPriorKnowledge() // #http2ClientWithPriorKnowledge diff --git a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/http2-client-alpn-fallback.excludes b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/http2-client-alpn-fallback.excludes new file mode 100644 index 0000000000..08394d6104 --- /dev/null +++ b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/http2-client-alpn-fallback.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. + +# new HTTP/2 client connection builder method that falls back to HTTP/1.1 over ALPN +# both traits are @DoNotInherit +ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.OutgoingConnectionBuilder.http2WithFallback") +ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.OutgoingConnectionBuilder.http2WithFallback") diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/AlpnObservingSSLEngine.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/AlpnObservingSSLEngine.scala new file mode 100644 index 0000000000..3c6bce9317 --- /dev/null +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/AlpnObservingSSLEngine.scala @@ -0,0 +1,119 @@ +/* + * 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. + */ + +package org.apache.pekko.http.impl.engine.http2 + +import org.apache.pekko +import pekko.annotation.InternalApi + +import java.nio.ByteBuffer +import java.util.function.BiFunction +import java.{ util => ju } +import javax.net.ssl.{ SSLEngine, SSLEngineResult, SSLParameters, SSLSession } + +/** + * INTERNAL API + * + * Delegating [[SSLEngine]] that reports the ALPN protocol as soon as the handshake has settled it. + * + * The JDK exposes the client-side ALPN result only through [[SSLEngine#getApplicationProtocol]], and neither + * `SSLEngine` nor Pekko's TLS stage offers an event for "handshake complete". `setHandshakeApplicationProtocolSelector` + * is only meaningful for the peer that selects the protocol (the server), and `HandshakeCompletedListener` exists on + * `SSLSocket` only. So the only portable place to observe the transition is around `wrap`/`unwrap`: once either has run + * far enough for the handshake to complete, `getApplicationProtocol` stops returning `null`. + * + * `onNegotiated` is invoked at most once, from whichever thread the TLS stage runs `wrap`/`unwrap` on. It is passed the + * empty string when the peer did not negotiate any protocol, which is what the JDK reports for a server that does not + * speak ALPN. + */ +@InternalApi +private[http] final class AlpnObservingSSLEngine(delegate: SSLEngine, onNegotiated: String => Unit) + extends SSLEngine(delegate.getPeerHost, delegate.getPeerPort) { + + // only ever touched from the TLS stage, whose calls into the engine are serialized + private[this] var reported = false + + private def observe(): Unit = + if (!reported) { + val protocol = + try delegate.getApplicationProtocol + catch { + // engines predating JDK 9 (or custom ones) may not implement it; treat as "no protocol negotiated" + case _: UnsupportedOperationException => "" + } + if (protocol ne null) { + reported = true + onNegotiated(protocol) + } + } + + override def wrap(srcs: Array[ByteBuffer], offset: Int, length: Int, dst: ByteBuffer): SSLEngineResult = { + val result = delegate.wrap(srcs, offset, length, dst) + observe() + result + } + + override def unwrap(src: ByteBuffer, dsts: Array[ByteBuffer], offset: Int, length: Int): SSLEngineResult = { + val result = delegate.unwrap(src, dsts, offset, length) + observe() + result + } + + override def getDelegatedTask: Runnable = delegate.getDelegatedTask + + override def closeInbound(): Unit = delegate.closeInbound() + override def isInboundDone: Boolean = delegate.isInboundDone + override def closeOutbound(): Unit = delegate.closeOutbound() + override def isOutboundDone: Boolean = delegate.isOutboundDone + + override def getSupportedCipherSuites: Array[String] = delegate.getSupportedCipherSuites + override def getEnabledCipherSuites: Array[String] = delegate.getEnabledCipherSuites + override def setEnabledCipherSuites(suites: Array[String]): Unit = delegate.setEnabledCipherSuites(suites) + + override def getSupportedProtocols: Array[String] = delegate.getSupportedProtocols + override def getEnabledProtocols: Array[String] = delegate.getEnabledProtocols + override def setEnabledProtocols(protocols: Array[String]): Unit = delegate.setEnabledProtocols(protocols) + + override def getSession: SSLSession = delegate.getSession + override def getHandshakeSession: SSLSession = delegate.getHandshakeSession + + override def beginHandshake(): Unit = delegate.beginHandshake() + override def getHandshakeStatus: SSLEngineResult.HandshakeStatus = delegate.getHandshakeStatus + + override def setUseClientMode(mode: Boolean): Unit = delegate.setUseClientMode(mode) + override def getUseClientMode: Boolean = delegate.getUseClientMode + override def setNeedClientAuth(need: Boolean): Unit = delegate.setNeedClientAuth(need) + override def getNeedClientAuth: Boolean = delegate.getNeedClientAuth + override def setWantClientAuth(want: Boolean): Unit = delegate.setWantClientAuth(want) + override def getWantClientAuth: Boolean = delegate.getWantClientAuth + + override def setEnableSessionCreation(flag: Boolean): Unit = delegate.setEnableSessionCreation(flag) + override def getEnableSessionCreation: Boolean = delegate.getEnableSessionCreation + + override def getSSLParameters: SSLParameters = delegate.getSSLParameters + override def setSSLParameters(params: SSLParameters): Unit = delegate.setSSLParameters(params) + + override def getApplicationProtocol: String = delegate.getApplicationProtocol + override def getHandshakeApplicationProtocol: String = delegate.getHandshakeApplicationProtocol + + override def setHandshakeApplicationProtocolSelector( + selector: BiFunction[SSLEngine, ju.List[String], String]): Unit = + delegate.setHandshakeApplicationProtocolSelector(selector) + + override def getHandshakeApplicationProtocolSelector: BiFunction[SSLEngine, ju.List[String], String] = + delegate.getHandshakeApplicationProtocolSelector +} diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/ClientProtocolSwitch.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/ClientProtocolSwitch.scala new file mode 100644 index 0000000000..76551cb188 --- /dev/null +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/ClientProtocolSwitch.scala @@ -0,0 +1,193 @@ +/* + * 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. + */ + +package org.apache.pekko.http.impl.engine.http2 + +import org.apache.pekko +import pekko.NotUsed +import pekko.annotation.InternalApi +import pekko.http.scaladsl.model.{ HttpRequest, HttpResponse } +import pekko.stream.ActorAttributes.Dispatcher +import pekko.stream.TLSProtocol.{ SslTlsInbound, SslTlsOutbound } +import pekko.stream.scaladsl.{ BidiFlow, Flow, Sink, Source } +import pekko.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler } +import pekko.stream.{ Attributes, BidiShape, Inlet, Outlet } + +import javax.net.ssl.SSLException +import scala.concurrent.{ ExecutionContext, Future } +import scala.util.{ Failure, Success, Try } + +/** + * INTERNAL API + * + * Client-side counterpart of [[ProtocolSwitch]]: installs either the HTTP/1.1 or the HTTP/2 client layer once ALPN + * has settled which protocol the server speaks. + * + * [[ProtocolSwitch]] can decide on the first inbound `SessionBytes`, because server-side the client always speaks + * first. That trigger deadlocks here: when negotiation lands on HTTP/1.1 the server stays silent until it receives a + * request, and the switch would be holding that request while waiting for inbound bytes. So the decision is instead + * driven by `negotiatedProtocol`, which [[AlpnObservingSSLEngine]] completes from the TLS handshake itself. + * + * Because that future is completed on the TLS stage's thread and delivered through an `AsyncCallback`, inbound + * elements can overtake it — an HTTP/2 server sends its SETTINGS frame immediately after the handshake. Those are + * buffered and replayed into the installed layer. + */ +@InternalApi +private[http] object ClientProtocolSwitch { + type ClientLayer = BidiFlow[HttpRequest, SslTlsOutbound, SslTlsInbound, HttpResponse, NotUsed] + + def apply(negotiatedProtocol: Future[String], http1: ClientLayer, http2: ClientLayer): ClientLayer = + BidiFlow.fromGraph(new GraphStage[BidiShape[HttpRequest, SslTlsOutbound, SslTlsInbound, HttpResponse]] { + val appIn = Inlet[HttpRequest]("ClientProtocolSwitch.appIn") + val netOut = Outlet[SslTlsOutbound]("ClientProtocolSwitch.netOut") + val netIn = Inlet[SslTlsInbound]("ClientProtocolSwitch.netIn") + val appOut = Outlet[HttpResponse]("ClientProtocolSwitch.appOut") + + override val shape: BidiShape[HttpRequest, SslTlsOutbound, SslTlsInbound, HttpResponse] = + BidiShape(appIn, netOut, netIn, appOut) + + override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) { + private var installed = false + private var appInFinished = false + private var pendingInbound: Vector[SslTlsInbound] = Vector.empty + + override def preStart(): Unit = { + // give TLS somewhere to put decrypted bytes; nothing is emitted until the handshake is through + pull(netIn) + + val callback = getAsyncCallback[Try[String]] { + case Success(protocol) => + if (!installed) install(if (protocol == Http2AlpnSupport.H2) http2 else http1) + case Failure(cause) => failStage(cause) + } + negotiatedProtocol.onComplete(callback.invoke)(ExecutionContext.parasitic) + } + + // -- handlers used until a layer is installed -- + + setHandler(appIn, + new InHandler { + // appIn is never pulled before a layer is installed, so onPush cannot fire + override def onPush(): Unit = () + override def onUpstreamFinish(): Unit = appInFinished = true + override def onUpstreamFailure(cause: Throwable): Unit = failStage(cause) + }) + + setHandler(netIn, + new InHandler { + // buffer and stop pulling: whatever arrives here is already past the handshake + override def onPush(): Unit = pendingInbound :+= grab(netIn) + override def onUpstreamFinish(): Unit = + failStage(new SSLException("Connection closed before the ALPN protocol was negotiated")) + override def onUpstreamFailure(cause: Throwable): Unit = failStage(cause) + }) + + setHandler(netOut, GraphStageLogic.EagerTerminateOutput) + setHandler(appOut, GraphStageLogic.EagerTerminateOutput) + + def install(layer: ClientLayer): Unit = { + installed = true + + val appDataOut = new SubSourceOutlet[HttpRequest]("ClientProtocolSwitch.appDataOut") + val appDataIn = new SubSinkInlet[HttpResponse]("ClientProtocolSwitch.appDataIn") + val netDataOut = new SubSourceOutlet[SslTlsInbound]("ClientProtocolSwitch.netDataOut") + val netDataIn = new SubSinkInlet[SslTlsOutbound]("ClientProtocolSwitch.netDataIn") + + val replay = pendingInbound + pendingInbound = Vector.empty + + connectIn(appIn, appDataOut, Vector.empty, appInFinished) + connectIn(netIn, netDataOut, replay, upstreamFinished = false) + connectOut(appDataIn, appOut) + connectOut(netDataIn, netOut) + + val attrs = + Attributes( + // don't (re)set dispatcher attribute to avoid adding an explicit async boundary + // between low-level and high-level stages + inheritedAttributes.attributeList.filterNot(_.isInstanceOf[Dispatcher])) + + Source.fromGraph(appDataOut.source) + .via(layer.addAttributes(attrs).join(Flow.fromSinkAndSource(netDataIn.sink, netDataOut.source))) + .runWith(Sink.fromGraph(appDataIn.sink))(interpreter.subFusingMaterializer) + } + + /** + * Feeds an outer inlet into the installed layer, replaying anything buffered while waiting for the + * negotiation result. + * + * `netIn` may still have the pull from `preStart` outstanding here, and an element may arrive for it before + * the sub-stream has any demand, so this both guards against pulling twice and buffers what it cannot push + * yet. Upstream completion is forwarded to the sub-stream only: the connection has to keep running so that + * in-flight responses can still be delivered. + */ + def connectIn[T](in: Inlet[T], out: SubSourceOutlet[T], buffered: Vector[T], upstreamFinished: Boolean) + : Unit = { + var pending = buffered + var finished = upstreamFinished + + def pump(): Unit = { + while (pending.nonEmpty && out.isAvailable) { + out.push(pending.head) + pending = pending.tail + } + if (pending.isEmpty) + if (finished) out.complete() + else if (out.isAvailable && !hasBeenPulled(in) && !isClosed(in)) pull(in) + } + + out.setHandler(new OutHandler { + override def onPull(): Unit = pump() + override def onDownstreamFinish(cause: Throwable): Unit = if (!isClosed(in)) cancel(in) + }) + + setHandler(in, + new InHandler { + override def onPush(): Unit = { + pending :+= grab(in) + pump() + } + override def onUpstreamFinish(): Unit = { + finished = true + if (pending.isEmpty) out.complete() + } + override def onUpstreamFailure(cause: Throwable): Unit = out.fail(cause) + }) + + pump() + } + + /** Drains the installed layer into an outer outlet. */ + def connectOut[T](in: SubSinkInlet[T], out: Outlet[T]): Unit = { + in.setHandler(new InHandler { + override def onPush(): Unit = push(out, in.grab()) + override def onUpstreamFinish(): Unit = complete(out) + override def onUpstreamFailure(cause: Throwable): Unit = fail(out, cause) + }) + + setHandler(out, + new OutHandler { + override def onPull(): Unit = in.pull() + override def onDownstreamFinish(cause: Throwable): Unit = in.cancel() + }) + + // demand may already have arrived on the outer port while we were waiting for the handshake + if (isAvailable(out)) in.pull() + } + } + }) +} diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala index e4d2b3879a..d85c9d3be4 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala @@ -36,7 +36,7 @@ import pekko.http.scaladsl.Http.OutgoingConnection import pekko.http.scaladsl.{ ConnectionContext, Http, HttpsConnectionContext } import pekko.http.scaladsl.Http.ServerBinding import pekko.http.scaladsl.model._ -import pekko.http.scaladsl.model.headers.{ Connection, RawHeader, Upgrade, UpgradeProtocol } +import pekko.http.scaladsl.model.headers.{ Connection, Host, RawHeader, Upgrade, UpgradeProtocol } import pekko.http.scaladsl.model.http2.Http2SettingsHeader import pekko.http.scaladsl.settings.ClientConnectionSettings import pekko.http.scaladsl.settings.ServerSettings @@ -50,7 +50,7 @@ import pekko.Done import javax.net.ssl.SSLEngine import scala.collection.immutable -import scala.concurrent.{ ExecutionContext, Future } +import scala.concurrent.{ ExecutionContext, Future, Promise } import scala.concurrent.duration.Duration import scala.util.control.NonFatal import scala.util.{ Failure, Success } @@ -288,6 +288,47 @@ private[http] final class Http2Ext(implicit val system: ActorSystem) .addAttributes(Http.cancellationStrategyAttributeForDelay(clientConnectionSettings.streamCancellationDelay)) } + /** + * Like [[outgoingConnection]], but offers `http/1.1` alongside `h2` over ALPN and falls back to the HTTP/1.1 client + * stack when the server does not select `h2`. + * + * The whole stack is built inside `Flow.fromMaterializer` so that every materialization gets its own engine and + * negotiation promise - `PersistentConnection.managedConnection` re-materializes the connection flow on each + * reconnect, so the "not reusable" approach `httpsWithAlpn` takes server-side would not work here. + */ + def outgoingConnectionWithNegotiation(host: String, port: Int, connectionContext: HttpsConnectionContext, + clientConnectionSettings: ClientConnectionSettings, log: LoggingAdapter) + : Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] = + Flow.fromMaterializer { (_, _) => + val negotiated = Promise[String]() + + // TODO find an alternative way to do this + def createEngine(): SSLEngine = { + val engine = connectionContext.engineCreator(Some((host, port))) + engine.setUseClientMode(true) + Http2AlpnSupport.clientSetApplicationProtocols(engine, Array(Http2AlpnSupport.H2, Http2AlpnSupport.HTTP11)) + new AlpnObservingSSLEngine(engine, protocol => { negotiated.trySuccess(protocol); () }) + } + + val hostHeader = port match { + case 0 | 443 => Host(host) + case _ => Host(host, port) + } + val http1Layer = http.clientLayer(hostHeader, clientConnectionSettings, log) + val http2Layer = + Http2Blueprint.clientStack(clientConnectionSettings, log, telemetry).atop(Http2Blueprint.unwrapTls) + + val stack = ClientProtocolSwitch(negotiated.future, http1Layer, http2Layer).addAttributes( + prepareClientAttributes(host, port)).atop( + LogByteStringTools.logTLSBidiBySetting("client-plain-text", + clientConnectionSettings.logUnencryptedNetworkBytes)).atop( + TLS(createEngine _, closing = TLSClosing.eagerClose)) + + stack.joinMat(clientConnectionSettings.transport.connectTo(host, port, clientConnectionSettings)( + system.classicSystem))(Keep.right) + .addAttributes(Http.cancellationStrategyAttributeForDelay(clientConnectionSettings.streamCancellationDelay)) + }.mapMaterializedValue(_.flatten) + def outgoingConnectionPriorKnowledge(host: String, port: Int, clientConnectionSettings: ClientConnectionSettings, log: LoggingAdapter): Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] = { val stack = Http2Blueprint.clientStack(clientConnectionSettings, log, telemetry).addAttributes( diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2AlpnSupport.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2AlpnSupport.scala index e8cecbfc8b..eeeb1cfa73 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2AlpnSupport.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2AlpnSupport.scala @@ -73,7 +73,7 @@ private[http] object Http2JDKAlpnSupport { def clientSetApplicationProtocols(engine: SSLEngine, protocols: Array[String]): Unit = { val params = engine.getSSLParameters - params.setApplicationProtocols(Array("h2")) + params.setApplicationProtocols(protocols) engine.setSSLParameters(params) } } diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/OutgoingConnectionBuilderImpl.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/OutgoingConnectionBuilderImpl.scala index 2eb3f69ed3..c4eaf448e0 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/OutgoingConnectionBuilderImpl.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/OutgoingConnectionBuilderImpl.scala @@ -94,6 +94,14 @@ private[pekko] object OutgoingConnectionBuilderImpl { log) } + override def http2WithFallback(): Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] = { + // tls, http/2 if the server selects it over ALPN, http/1.1 otherwise + val port = this.port.getOrElse(443) + Http2(system.classicSystem).outgoingConnectionWithNegotiation(host, port, + connectionContext.getOrElse(Http(system.classicSystem).defaultClientHttpsContext), clientConnectionSettings, + log) + } + override def managedPersistentHttp2(): Flow[HttpRequest, HttpResponse, NotUsed] = PersistentConnection.managedConnection( http2(), @@ -128,6 +136,10 @@ private[pekko] object OutgoingConnectionBuilderImpl { : JFlow[javadsl.model.HttpRequest, javadsl.model.HttpResponse, CompletionStage[javadsl.OutgoingConnection]] = javaFlow(actual.https()) + override def http2WithFallback() + : JFlow[javadsl.model.HttpRequest, javadsl.model.HttpResponse, CompletionStage[javadsl.OutgoingConnection]] = + javaFlow(actual.http2WithFallback()) + override def managedPersistentHttp2(): JFlow[javadsl.model.HttpRequest, javadsl.model.HttpResponse, NotUsed] = javaFlowKeepMatVal(actual.managedPersistentHttp2()) diff --git a/http-core/src/main/scala/org/apache/pekko/http/javadsl/OutgoingConnectionBuilder.scala b/http-core/src/main/scala/org/apache/pekko/http/javadsl/OutgoingConnectionBuilder.scala index 0d407ddbcb..a10fc7fcbd 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/javadsl/OutgoingConnectionBuilder.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/javadsl/OutgoingConnectionBuilder.scala @@ -64,6 +64,18 @@ trait OutgoingConnectionBuilder { */ def http2(): Flow[HttpRequest, HttpResponse, CompletionStage[OutgoingConnection]] + /** + * Create a flow that when materialized creates a single TLS connection with a default port 443, offering both + * `h2` and `http/1.1` over ALPN and using whichever the server selects. Unlike `http2()` this does not fail + * against a server that has no HTTP/2 support - it falls back to HTTP/1.1. + * + * Note that when the negotiation ends up on HTTP/2 the responses are not guaranteed to arrive in the same order as + * the requests go out, so requests need a [[pekko.http.javadsl.model.RequestResponseAssociation]] + * which Pekko HTTP will carry over to the corresponding response for a request. + */ + @ApiMayChange + def http2WithFallback(): Flow[HttpRequest, HttpResponse, CompletionStage[OutgoingConnection]] + /** * Create a flow that when materialized creates a managed HTTP/2 TLS connection with a default port 443. * diff --git a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/OutgoingConnectionBuilder.scala b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/OutgoingConnectionBuilder.scala index 7bd0573339..1ee83631de 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/OutgoingConnectionBuilder.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/OutgoingConnectionBuilder.scala @@ -65,6 +65,18 @@ trait OutgoingConnectionBuilder { */ def http2(): Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] + /** + * Create a flow that when materialized creates a single TLS connection with a default port 443, offering both + * `h2` and `http/1.1` over ALPN and using whichever the server selects. Unlike `http2()` this does not fail + * against a server that has no HTTP/2 support - it falls back to HTTP/1.1. + * + * Note that when the negotiation ends up on HTTP/2 the responses are not guaranteed to arrive in the same order as + * the requests go out, so requests need a [[pekko.http.scaladsl.model.RequestResponseAssociation]] + * which Pekko HTTP will carry over to the corresponding response for a request. + */ + @ApiMayChange + def http2WithFallback(): Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] + /** * Create a flow that when materialized creates a managed HTTP/2 TLS connection with a default port 443. * diff --git a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientFallbackSpec.scala b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientFallbackSpec.scala new file mode 100644 index 0000000000..fa7b6702fd --- /dev/null +++ b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientFallbackSpec.scala @@ -0,0 +1,112 @@ +/* + * 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. + */ + +package org.apache.pekko.http.impl.engine.http2 + +import org.apache.pekko +import pekko.http.impl.util.{ ExampleHttpContexts, PekkoSpecWithMaterializer } +import pekko.http.scaladsl.Http +import pekko.http.scaladsl.model.{ + AttributeKey, + HttpProtocols, + HttpRequest, + HttpResponse, + RequestResponseAssociation, + StatusCodes +} +import pekko.http.scaladsl.settings.{ ClientConnectionSettings, ServerSettings } +import pekko.http.scaladsl.unmarshalling.Unmarshal +import pekko.stream.scaladsl.{ Sink, Source } +import pekko.stream.testkit.{ TestPublisher, TestSubscriber } +import pekko.testkit.TestProbe + +import scala.concurrent.Future +import org.scalatest.concurrent.ScalaFutures + +/** + * Covers `OutgoingConnectionBuilder.http2WithFallback`, which offers both `h2` and `http/1.1` over ALPN and picks + * the client stack from what the server selected. + */ +class Http2ClientFallbackSpec extends PekkoSpecWithMaterializer(""" + pekko.http.server.log-unencrypted-network-bytes = 100 + pekko.http.client.log-unencrypted-network-bytes = 100 + pekko.actor.serialize-messages = false + """) with ScalaFutures { + + case class RequestId(id: String) extends RequestResponseAssociation + val requestIdAttr = AttributeKey[RequestId]("requestId") + + "The HTTP/2 client with ALPN fallback" should { + + "negotiate HTTP/2 against a server that supports it" in new TestSetup { + val response = roundTrip() + response.status shouldBe StatusCodes.OK + Unmarshal(response.entity).to[String].futureValue shouldBe "pong" + + // the stream id attribute is only ever set by the HTTP/2 server stack + serverSeenRequest.attribute(Http2.streamId) shouldBe Symbol("nonEmpty") + serverSeenRequest.protocol shouldBe HttpProtocols.`HTTP/2.0` + } + + "fall back to HTTP/1.1 against a server that does not support HTTP/2" in new TestSetup { + override def serverSettings: ServerSettings = super.serverSettings.withEnableHttp2(false) + + val response = roundTrip() + response.status shouldBe StatusCodes.OK + Unmarshal(response.entity).to[String].futureValue shouldBe "pong" + + serverSeenRequest.attribute(Http2.streamId) shouldBe Symbol("empty") + serverSeenRequest.protocol shouldBe HttpProtocols.`HTTP/1.1` + } + } + + class TestSetup { + def serverSettings: ServerSettings = ServerSettings(system) + def clientSettings: ClientConnectionSettings = ClientConnectionSettings(system) + + private val serverRequestProbe = TestProbe() + + lazy val binding = + Http().newServerAt("localhost", 0) + .enableHttps(ExampleHttpContexts.exampleServerContext) + .withSettings(serverSettings) + .bind { request => + serverRequestProbe.ref ! request + Future.successful(HttpResponse(entity = "pong")) + }.futureValue + + lazy val clientFlow = + Http().connectionTo("pekko.example.org") + .withCustomHttpsConnectionContext(ExampleHttpContexts.exampleClientContext) + .withClientConnectionSettings( + clientSettings.withTransport(ExampleHttpContexts.proxyTransport(binding.localAddress))) + .http2WithFallback() + + lazy val requestsOut = TestPublisher.probe[HttpRequest]() + lazy val responsesIn = TestSubscriber.probe[HttpResponse]() + Source.fromPublisher(requestsOut) + .via(clientFlow) + .runWith(Sink.fromSubscriber(responsesIn)) + + def roundTrip(): HttpResponse = { + requestsOut.sendNext(HttpRequest().addAttribute(requestIdAttr, RequestId("request-1"))) + responsesIn.requestNext() + } + + lazy val serverSeenRequest: HttpRequest = serverRequestProbe.expectMsgType[HttpRequest] + } +} From 054abfcca5d9af935b34a2fd33a0df5ea5e94748 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Sun, 30 Aug 2026 12:45:44 +0100 Subject: [PATCH 2/3] http/2: avoid buffering every element on the negotiated connection connectIn ran every element through the replay buffer, allocating a Vector append on push and a tail on drain for the whole life of the connection. The buffer only exists to hold elements that overtake the negotiation callback, so push straight to the sub-source when nothing is pending and demand is there. Co-Authored-By: Claude Opus 5 (1M context) --- .../http/impl/engine/http2/ClientProtocolSwitch.scala | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/ClientProtocolSwitch.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/ClientProtocolSwitch.scala index 76551cb188..54de951978 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/ClientProtocolSwitch.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/ClientProtocolSwitch.scala @@ -158,8 +158,14 @@ private[http] object ClientProtocolSwitch { setHandler(in, new InHandler { override def onPush(): Unit = { - pending :+= grab(in) - pump() + val elem = grab(in) + // steady state: hand the element straight over, the buffer is only needed while the negotiation + // result is still in flight + if (pending.isEmpty && out.isAvailable) out.push(elem) + else { + pending :+= elem + pump() + } } override def onUpstreamFinish(): Unit = { finished = true From 0bd5ffd8d9701f03972995503674b3ebdc07ddb8 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Sun, 30 Aug 2026 13:23:20 +0100 Subject: [PATCH 3/3] docs: list http2WithFallback in the compatibility guidelines ApiMayChangeDocCheckerSpec scans for @ApiMayChange members and requires each to be named in compatibility-guidelines.md, so adding the annotated method to both OutgoingConnectionBuilder traits broke it. sbt "docs/testOnly docs.ApiMayChangeDocCheckerSpec" - 2 passed Co-Authored-By: Claude Opus 5 (1M context) --- docs/src/main/paradox/compatibility-guidelines.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/src/main/paradox/compatibility-guidelines.md b/docs/src/main/paradox/compatibility-guidelines.md index 9f37c8b4cb..e00e3a7255 100644 --- a/docs/src/main/paradox/compatibility-guidelines.md +++ b/docs/src/main/paradox/compatibility-guidelines.md @@ -28,6 +28,7 @@ Scala org.apache.pekko.http.scaladsl.unmarshalling.sse.EventStreamUnmarshalling org.apache.pekko.http.scaladsl.OutgoingConnectionBuilder#managedPersistentHttp2 org.apache.pekko.http.scaladsl.OutgoingConnectionBuilder#managedPersistentHttp2WithPriorKnowledge + org.apache.pekko.http.scaladsl.OutgoingConnectionBuilder#http2WithFallback ``` Java @@ -41,6 +42,7 @@ Java org.apache.pekko.http.javadsl.model.RequestResponseAssociation org.apache.pekko.http.javadsl.OutgoingConnectionBuilder#managedPersistentHttp2WithPriorKnowledge org.apache.pekko.http.javadsl.OutgoingConnectionBuilder#managedPersistentHttp2 + org.apache.pekko.http.javadsl.OutgoingConnectionBuilder#http2WithFallback ``` #### pekko-http-caching