http/2: prototype ALPN fallback from HTTP/2 to HTTP/1.1 for client connections

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

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

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

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

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

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
+}