forked from akka/akka-http
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
=htc Support
UseHttp2.Negotiated for http2 connections over plain h…
…ttp (akka#2543) By looking ahead and deciding on the prior knowledge preface a potential HTTP/2 client sends whether to use HTTP/1.1 or HTTP/2. Co-Authored-By: Arnout Engelen <github@bzzt.net> Co-Authored-By: Johannes Rudolph <johannes.rudolph@gmail.com>
- Loading branch information
1 parent
9c3274a
commit c00ec4c
Showing
9 changed files
with
312 additions
and
51 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
141 changes: 141 additions & 0 deletions
141
akka-http2-support/src/main/scala/akka/http/impl/engine/http2/PriorKnowledgeSwitch.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,141 @@ | ||
| /* | ||
| * Copyright (C) 2009-2019 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.http.impl.engine.http2 | ||
|
|
||
| import javax.net.ssl.SSLException | ||
|
|
||
| import akka.util.ByteString | ||
| import akka.NotUsed | ||
| import akka.annotation.InternalApi | ||
| import akka.http.impl.engine.server.HttpAttributes | ||
| import akka.http.scaladsl.model.{ HttpRequest, HttpResponse } | ||
| import akka.stream.TLSProtocol.{ SessionBytes, SessionTruncated, SslTlsInbound, SslTlsOutbound } | ||
| import akka.stream.scaladsl.Flow | ||
| import akka.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler } | ||
| import akka.stream._ | ||
|
|
||
| /** INTERNAL API */ | ||
| @InternalApi | ||
| private[http] object PriorKnowledgeSwitch { | ||
| type HttpServerFlow = Flow[ByteString, ByteString, NotUsed] | ||
| type HttpServerShape = FlowShape[ByteString, ByteString] | ||
|
|
||
| private final val HTTP2_CONNECTION_PREFACE = ByteString("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n") | ||
|
|
||
| def apply( | ||
| http1Stack: HttpServerFlow, | ||
| http2Stack: HttpServerFlow): HttpServerFlow = | ||
| Flow.fromGraph( | ||
| new GraphStage[HttpServerShape] { | ||
|
|
||
| // --- outer ports --- | ||
| val netIn = Inlet[ByteString]("PriorKnowledgeSwitch.netIn") | ||
| val netOut = Outlet[ByteString]("PriorKnowledgeSwitch.netOut") | ||
| // --- end of outer ports --- | ||
|
|
||
| override val shape: HttpServerShape = | ||
| FlowShape(netIn, netOut) | ||
|
|
||
| override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) { | ||
| logic => | ||
|
|
||
| // --- inner ports, bound to actual server in install call --- | ||
| val serverDataIn = new SubSinkInlet[ByteString]("ServerImpl.netIn") | ||
| val serverDataOut = new SubSourceOutlet[ByteString]("ServerImpl.netOut") | ||
| // --- end of inner ports --- | ||
|
|
||
| override def preStart(): Unit = pull(netIn) | ||
|
|
||
| setHandler(netIn, new InHandler { | ||
| private[this] var grabbed = ByteString.empty | ||
| def onPush(): Unit = { | ||
| val data = grabbed ++ grab(netIn) | ||
| if (data.length >= HTTP2_CONNECTION_PREFACE.length) { // We should know by now | ||
| if (data.startsWith(HTTP2_CONNECTION_PREFACE, 0)) | ||
| install(http2Stack, data) | ||
| else | ||
| install(http1Stack, data) | ||
| } else if (data.isEmpty || data.startsWith(HTTP2_CONNECTION_PREFACE, 0)) { // Still unknown | ||
| grabbed = data | ||
| } else { // Not a Prior Knowledge request | ||
| install(http1Stack, data) | ||
| } | ||
| } | ||
| }) | ||
|
|
||
| setHandler(netOut, new OutHandler { def onPull(): Unit = () }) // Ignore pull | ||
|
|
||
| def install(serverImplementation: HttpServerFlow, firstElement: ByteString): Unit = { | ||
| connect(netIn, serverDataOut, Some(firstElement)) | ||
| connect(serverDataIn, netOut) | ||
|
|
||
| serverImplementation | ||
| .addAttributes(inheritedAttributes) // propagate attributes to "real" server (such as HttpAttributes) | ||
| .join(Flow.fromSinkAndSource(serverDataIn.sink, serverDataOut.source)) // Network side | ||
| .run()(interpreter.subFusingMaterializer) | ||
| } | ||
|
|
||
| // helpers to connect inlets and outlets also binding completion signals of given ports | ||
| def connect[T](in: Inlet[T], out: SubSourceOutlet[T], initialElement: Option[T]): Unit = { | ||
|
|
||
| val firstElementHandler = { | ||
| val propagatePull = new OutHandler { override def onPull(): Unit = pull(in) } | ||
|
|
||
| initialElement match { | ||
| case Some(ele) if out.isAvailable => | ||
| out.push(ele) | ||
| propagatePull | ||
| case Some(ele) => | ||
| new OutHandler { | ||
| override def onPull(): Unit = { | ||
| out.push(ele) | ||
| out.setHandler(propagatePull) | ||
| } | ||
| } | ||
| case None => propagatePull | ||
| } | ||
| } | ||
|
|
||
| out.setHandler(firstElementHandler) | ||
|
|
||
| setHandler(in, new InHandler { | ||
| override def onPush(): Unit = out.push(grab(in)) | ||
|
|
||
| override def onUpstreamFinish(): Unit = { | ||
| out.complete() | ||
| super.onUpstreamFinish() | ||
| } | ||
|
|
||
| override def onUpstreamFailure(ex: Throwable): Unit = { | ||
| out.fail(ex) | ||
| super.onUpstreamFailure(ex) | ||
| } | ||
| }) | ||
|
|
||
| if (out.isAvailable) pull(in) // to account for lost pulls during initialization | ||
| } | ||
|
|
||
| def connect[T](in: SubSinkInlet[T], out: Outlet[T]): Unit = { | ||
| val handler = new InHandler { | ||
| override def onPush(): Unit = push(out, in.grab()) | ||
| } | ||
|
|
||
| val outHandler = new OutHandler { | ||
| override def onPull(): Unit = in.pull() | ||
| override def onDownstreamFinish(): Unit = { | ||
| in.cancel() | ||
| super.onDownstreamFinish() | ||
| } | ||
| } | ||
|
|
||
| in.setHandler(handler) | ||
| setHandler(out, outHandler) | ||
|
|
||
| if (isAvailable(out)) in.pull() // to account for lost pulls during initialization | ||
| } | ||
| } | ||
| } | ||
| ) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
82 changes: 82 additions & 0 deletions
82
akka-http2-support/src/test/scala/akka/http/impl/engine/http2/WithPriorKnowledgeSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| /* | ||
| * Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.http.impl.engine.http2 | ||
|
|
||
| import akka.http.scaladsl.model.{ HttpResponse, StatusCodes } | ||
| import akka.http.scaladsl.{ Http, HttpConnectionContext, UseHttp2 } | ||
| import akka.http.scaladsl.model.{ HttpRequest, HttpResponse, HttpProtocols } | ||
| import akka.stream.ActorMaterializer | ||
| import java.util.Base64 | ||
| import akka.stream.scaladsl.{ Sink, Source, Tcp } | ||
| import akka.testkit.AkkaSpec | ||
| import akka.util.ByteString | ||
|
|
||
| import scala.concurrent.Future | ||
| import akka.stream.OverflowStrategy | ||
| import akka.stream.scaladsl.{ SinkQueue, Keep } | ||
|
|
||
| class WithPriorKnowledgeSpec extends AkkaSpec(""" | ||
| akka.loglevel = warning | ||
| akka.loggers = ["akka.testkit.TestEventListener"] | ||
| akka.http.server.preview.enable-http2 = on | ||
| """) { | ||
|
|
||
| implicit val ec = system.dispatcher | ||
|
|
||
| "An HTTP server with PriorKnowledge" should { | ||
| implicit val mat = ActorMaterializer() | ||
|
|
||
| val binding = Http().bindAndHandleAsync( | ||
| _ ⇒ Future.successful(HttpResponse(status = StatusCodes.ImATeapot)), | ||
| "127.0.0.1", | ||
| port = 0, | ||
| HttpConnectionContext(UseHttp2.Negotiated) | ||
| ).futureValue | ||
|
|
||
| "respond to cleartext HTTP/1.1 requests with cleartext HTTP/1.1" in { | ||
| val (host, port) = (binding.localAddress.getHostName, binding.localAddress.getPort) | ||
| val responseFuture: Future[HttpResponse] = Http().singleRequest(HttpRequest(uri = s"http://$host:$port")) | ||
| val response = responseFuture.futureValue | ||
| response.protocol should be(HttpProtocols.`HTTP/1.1`) | ||
| response.status should be(StatusCodes.ImATeapot) | ||
| } | ||
|
|
||
| "respond to cleartext HTTP/2 requests with cleartext HTTP/2" in { | ||
| val (host, port) = (binding.localAddress.getHostName, binding.localAddress.getPort) | ||
|
|
||
| val (source, sink) = | ||
| Source.queue[String](1000, OverflowStrategy.fail) | ||
| .map(str => ByteString(Base64.getDecoder.decode(str))) | ||
| .via(Tcp().outgoingConnection(host, port)) | ||
| .toMat(Sink.queue())(Keep.both) | ||
| .run() | ||
|
|
||
| // Obtained by converting the input request bytes from curl with --http2-prior-knowledge | ||
| // This includes port 9009 as 'authority', which our server accepts. | ||
| source.offer("UFJJICogSFRUUC8yLjANCg0KU00NCg0KAAASBAAAAAAAAAMAAABkAARAAAAAAAIAAAAAAAAECAAAAAAAP/8AAQAAHgEFAAAAAYKEhkGKCJ1cC4Fw3HwAf3qIJbZQw6u20uBTAyovKg==").futureValue | ||
|
|
||
| // read settings frame | ||
| Http2Protocol.FrameType.byId(sink.pull().futureValue.get(3)) should be(Http2Protocol.FrameType.SETTINGS) | ||
| // read settings frame | ||
| Http2Protocol.FrameType.byId(sink.pull().futureValue.get(3)) should be(Http2Protocol.FrameType.SETTINGS) | ||
| // ack settings | ||
| source.offer("AAAABAEAAAAA") | ||
|
|
||
| val response = readSink(sink).futureValue | ||
| val tpe = Http2Protocol.FrameType.byId(response(3)) | ||
| tpe should be(Http2Protocol.FrameType.HEADERS) | ||
| response.map(_.toChar).mkString should include("418") | ||
| } | ||
| } | ||
|
|
||
| private def readSink(sink: SinkQueue[ByteString]): Future[ByteString] = { | ||
| sink.pull().flatMap { | ||
| case Some(bytes) if bytes.isEmpty => | ||
| readSink(sink) | ||
| case Some(bytes) => | ||
| Future.successful(bytes) | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.