Programming paradigms

As explained here ServiceTalk supports multiple programming paradigms for users to use. In this document we elaborate on how HTTP module provides support for these programming paradigms.

Whenever a payload body is aggregated into a single object (the aggregated paradigms below, or any filter that calls toRequest()/toResponse()), the whole body is buffered in memory. To guard against excessive memory consumption, ServiceTalk limits the size of an aggregated body. The limit is measured against the fully decoded payload buffered at the point of aggregation — that is, after any decompression or body transformation applied by filters earlier in the pipeline — not the bytes received from the network nor any declared Content-Length. A small compressed body that expands beyond the limit once decoded is therefore still subject to it. Bodies consumed purely as a stream are not affected.

Configure this per client/server via HttpServerBuilder#maxAggregatedPayloadSize(int) / SingleAddressHttpClientBuilder#maxAggregatedPayloadSize(int), where the sign selects the mode and the magnitude selects the threshold: a positive value enforces the limit (the server responds 413 Payload Too Large; the client raises a PayloadTooLargeException), a negative value enables warn-only mode at its magnitude (oversized aggregated messages are still served/delivered, but a warning is logged, rate-limited per client/server), and 0 disables the limit. By default the client warns at 64 MiB and the server warns at 16 MiB; enforcing is planned to become the default for servers in a future release.

The built-in defaults can be overridden globally with the io.servicetalk.http.netty.defaultClientMaxAggregatedPayloadSize and io.servicetalk.http.netty.defaultServerMaxAggregatedPayloadSize system properties (same sign convention); a per-builder call takes precedence.

Server

An HTTP Server in ServiceTalk uses a user provided Service that implements all the business logic for that server. ServiceTalk allows users to write these Service implementations in different programming paradigms.

Blocking and Aggregated

This programming paradigm exposes the request (including the payload body) as a single aggregated object, and the response (including the payload body) is also a single aggregated object. The Service directly returns the response object and there are no asynchronous primitives involved.

See BlockingHelloWorldServer.

HttpServers.forPort(8080)
  .listenBlockingAndAwait((ctx, request, responseFactory) ->
    responseFactory.ok().payloadBody("Hello World!", textSerializer()))
  .awaitShutdown();

Blocking and Streaming

This programming paradigm exposes the request payload body as an Iterable of Buffers and the response payload body is written via an HttpPayloadWriter or an HttpOutputStream. There are no asynchronous primitives involved.

See BlockingHelloWorldStreamingServer.

HttpServers.forPort(8080).listenBlockingStreamingAndAwait((ctx, request, response) -> {
  try (HttpPayloadWriter<String> payloadWriter = response.sendMetaData(textSerializer())) {
    payloadWriter.write("Hello");
    payloadWriter.write(" World!");
  }
}).awaitShutdown();

Asynchronous and Aggregated

This programming paradigm exposes the request payload body as a single aggregated object, and the response payload body is also a single aggregated object. The Service provides the response by completing a Single which is an asynchronous primitive.

See HelloWorldServer.

HttpServers.forPort(8080)
  .listenAndAwait((ctx, request, responseFactory) ->
    succeeded(responseFactory.ok().payloadBody("Hello World!", textSerializer())))
  .awaitShutdown();

Asynchronous and Streaming

This programming paradigm exposes the request payload body as a Publisher typically of Buffers (although other types like file regions may be added), the response meta-data is provided by completing a Single, and the response payload body is written via a Publisher.

See HelloWorldStreamingServer.

HttpServers.forPort(8080)
  .listenStreamingAndAwait((ctx, request, responseFactory) ->
    succeeded(responseFactory.ok()
      .payloadBody(from("Hello", " World!"), textSerializer())))
  .awaitShutdown();

Client

An HTTP Client in ServiceTalk is used by users to send requests to a server. ServiceTalk allows users to create a Client for different programming paradigms.

Blocking and Aggregated

This programming paradigm expects the request (including the payload body) as a single aggregated object, and the response (including the payload body) is also a single aggregated object. The Client directly returns the response object and there are no asynchronous primitives involved.

See BlockingHelloWorldClient

try (BlockingHttpClient client = HttpClients.forSingleAddress("localhost", 8080).buildBlocking()) {
  HttpResponse response = client.request(client.get("/sayHello"));
  // use the response
}

Blocking and Streaming

This programming paradigm expects the request payload body as an Iterable of Buffers and the response payload body is consumed via an Iterable of Buffers. There are no asynchronous primitives involved.

See BlockingHelloWorldStreamingClient

try (BlockingStreamingHttpClient client = HttpClients.forSingleAddress("localhost", 8080)
      .buildBlockingStreaming()) {
  BlockingStreamingHttpResponse response = client.request(client.get("/sayHello"));
  System.out.println(response.toString((name, value) -> value));
  try (BlockingIterator<String> payload = response.payloadBody(textDeserializer()).iterator()) {
    while (payload.hasNext()) {
      System.out.println(payload.next());
    }
  }
}

Asynchronous and Aggregated

This programming paradigm expects the request payload body as a single aggregated object, and the response payload body is also a single aggregated object. The HttpClient provides the response via a Single which is an asynchronous primitive.

See HelloWorldClient

try (HttpClient client = HttpClients.forSingleAddress("localhost", 8080).build()) {
// This example is demonstrating asynchronous execution, but needs to prevent the main
// thread from exiting before the response has been processed. This isn't typical usage
// for a streaming API but is useful for demonstration purposes.
  CountDownLatch responseProcessedLatch = new CountDownLatch(1);
  client.request(client.get("/sayHello"))
        .doFinally(responseProcessedLatch::countDown)
        .subscribe(resp -> {
     System.out.println(resp.toString((name, value) -> value));
     System.out.println(resp.payloadBody(textDeserializer()));
  });

  responseProcessedLatch.await();
}

Asynchronous and Streaming

This programming paradigm expects the request payload body as a Publisher typically of Buffers (although other types like file regions may be added), the response meta-data is provided via a Single, and the response payload body is written via a Publisher.

See HelloWorldStreamingClient

try (StreamingHttpClient client = HttpClients.forSingleAddress("localhost", 8080).buildStreaming()) {
// This example is demonstrating asynchronous execution, but needs to prevent the main thread from
// exiting before the response has been processed. This isn't typical usage for a streaming API but
// is useful for demonstration purposes.
  CountDownLatch responseProcessedLatch = new CountDownLatch(1);
  client.request(client.get("/sayHello"))
        .beforeOnSuccess(response -> System.out.println(response.toString((name, value) -> value)))
        .flatMapPublisher(resp -> resp.payloadBody(textDeserializer()))
        .doFinally(responseProcessedLatch::countDown)
        .forEach(System.out::println);

  responseProcessedLatch.await();
}