Client-Side HTTP/2 (Preview)

Create the client

There are three mechanisms for a client to establish an HTTP/2 connection. Apache Pekko HTTP supports:

• HTTP/2 over TLS
• HTTP/2 over a plain TCP connection (“h2c with Prior Knowledge”)

Apache Pekko HTTP doesn’t support:

• HTTP Upgrade mechanism

HTTP/2 over TLS

To create a client, use the Http() fluent API to connect and use the http2() creator:

Scala

copysourceimport org.apache.pekko.http.scaladsl.Http

Http().connectionTo("localhost").toPort(8443).http2()

Java

copysourceimport org.apache.pekko.http.javadsl.Http;

Http.get(system).connectionTo("127.0.0.1").toPort(8443).http2();

HTTP/2 over TLS needs Application-Layer Protocol Negotiation (ALPN) 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.

h2c with Prior Knowledge

The other option is to connect and start communicating in HTTP/2 immediately. You must know beforehand the target server supports HTTP/2 over a plain TCP connection. For this reason this approach is known as h2c with Prior Knowledge of HTTP/2 support.

To create a client, use the Http() fluent API to connect and use the http2WithPriorKnowledge() creator:

Scala

copysourceimport org.apache.pekko.http.scaladsl.Http

Http().connectionTo("localhost").toPort(8080).http2WithPriorKnowledge()

Java

copysourceimport org.apache.pekko.http.javadsl.Http;

Http.get(system).connectionTo("127.0.0.1").toPort(8080).http2WithPriorKnowledge();

HTTP Upgrade mechanism

The Apache Pekko HTTP client doesn’t support HTTP/1 to HTTP/2 negotiation over plaintext using the Upgrade mechanism.

Limiting managed persistent connection lifetime

Managed persistent HTTP/2 clients can periodically retire long-lived connections so that later requests establish fresh connections. This is useful when clients would otherwise remain pinned to the same server instances after a scale-out or rolling deployment.

Configure the maximum age and optional jitter under the HTTP/2 client settings:

pekko.http.client.http2.persistent-connection-max-age = 10m
pekko.http.client.http2.persistent-connection-max-age-jitter = 0.1

The setting applies to managedPersistentHttp2() and managedPersistentHttp2WithPriorKnowledge(). Each connection gets an independently jittered age; with the default jitter of 0.1, retirement happens between 90% and 110% of the configured maximum age. Set the jitter to 0 to disable it.

Retirement is currently break-before-make. When a connection reaches its age, the managed client stops assigning new requests to it and lets requests already in flight drain before closing the connection. New requests are backpressured until the old connection disconnects, then a replacement connection is established. As a result, a retiring connection can add its drain time plus TCP/TLS/HTTP2 setup time to new-request latency.

The existing completion-timeout bounds the drain period. If that timeout expires, the old connection is closed even when requests are still in flight, terminating long-lived requests such as streaming responses.

The default maximum age is infinite, which disables age-based retirement. The settings can also be changed programmatically with Http2ClientSettings.withPersistentConnectionMaxAge and Http2ClientSettings.withPersistentConnectionMaxAgeJitter.

Request-response ordering

For HTTP/2 connections the responses are not guaranteed to arrive in the same order that the requests were emitted to the server, for example a request with a quickly available response may outrun a previous request that the server is slower to respond to. For HTTP/2 it is therefore important to have a way to correlate the response with the request it was made for. This can be achieved through a RequestResponseAssociationRequestResponseAssociation set on the request, Apache Pekko HTTP will pass such association objects on to the response.

In this sample the built-in org.apache.pekko.http.scaladsl.model.ResponsePromiseorg.apache.pekko.http.javadsl.model.ResponseFuture RequestResponseAssociation is used to return a FutureCompletionStage for the response:

Scala

copysource  val dispatch = singleRequest(Http().connectionTo("pekko.apache.org").http2())

  dispatch(
    HttpRequest(
      uri = "https://pekko.apache.org/api/pekko/current/org/apache/pekko/actor/typed/scaladsl/index.html",
      headers = headers.`Accept-Encoding`(HttpEncodings.gzip) :: Nil)).onComplete { res =>
    println(s"[1] Got index.html: $res")
    res.get.entity.dataBytes.runWith(Sink.ignore).onComplete(res => println(s"Finished reading [1] $res"))
  }

private def singleRequest(
    connection: Flow[HttpRequest, HttpResponse, Any], bufferSize: Int = 100)(implicit system: ActorSystem)
    : HttpRequest => Future[HttpResponse] = {
  val queue =
    Source.queue(bufferSize)
      .via(connection)
      .to(Sink.foreach { response =>
        // complete the response promise with the response when it arrives
        val responseAssociation = response.attribute(ResponsePromise.Key).get
        responseAssociation.promise.trySuccess(response)
      })
      .run()

  req => {
    // create a promise of the response for each request and set it as an attribute on the request
    val p = Promise[HttpResponse]()
    queue.offer(req.addAttribute(ResponsePromise.Key, ResponsePromise(p))) match {
      // return the future response
      case QueueOfferResult.Enqueued    => p.future
      case QueueOfferResult.Dropped     => Future.failed(new RuntimeException("Queue overflowed."))
      case QueueOfferResult.Failure(ex) => Future.failed(ex)
      case QueueOfferResult.QueueClosed => Future.failed(
          new RuntimeException("Queue was closed (pool shut down)."))
    }
  }
}

Java

copysource  Function<HttpRequest, CompletionStage<HttpResponse>> dispatch =
      singleRequest(system, Http.get(system).connectionTo("pekko.apache.org").http2());

  dispatch
      .apply(
          HttpRequest.create(
                  "https://pekko.apache.org/api/pekko/current/org/apache/pekko/actor/typed/scaladsl/index.html")
              .withHeaders(List.of(AcceptEncoding.create(HttpEncodings.GZIP))))
      .thenAccept(
          res -> {
            System.out.println("[1] Got index.html: " + res);
            res.entity()
                .getDataBytes()
                .runWith(Sink.ignore(), mat)
                .thenAccept(
                    consumedRes -> System.out.println("Finished reading [1] " + consumedRes));
          });

private static Function<HttpRequest, CompletionStage<HttpResponse>> singleRequest(
    ActorSystem system, Flow<HttpRequest, HttpResponse, ?> connection) {
  BoundedSourceQueue<HttpRequest> queue =
      Source.<HttpRequest>queue(100)
          .via(connection)
          .to(
              Sink.foreach(
                  res -> {
                    try {
                      // complete the future with the response when it arrives
                      ResponseFuture responseFuture =
                          res.getAttribute(ResponseFuture.KEY()).get();
                      responseFuture.future().complete(res);
                    } catch (Exception ex) {
                      system.log().error(ex, "Error processing HTTP/2 response");
                    }
                  }))
          .run(SystemMaterializer.get(system).materializer());

  return (HttpRequest req) -> {
    // create a future of the response for each request and set it as an attribute on the request
    CompletableFuture<HttpResponse> future = new CompletableFuture<>();
    ResponseFuture attribute = new ResponseFuture(future);
    queue.offer(req.addAttribute(ResponseFuture.KEY(), attribute));
    // return the future response
    return attribute.future();
  };
}