Improve timeout support for HTTP (#1456)

Motivation:
Some uses of the HTTP API require progress or completion within specified time limits.

Modifications:
New or enhanced filters are proivded for client and server to manage timeout events.
Existing client filter `TimeoutHttpRequesterFilter` is enhanced to support additional
timeout options for the entire HTTP request/response transaction. A new server filter
`TimeoutHttpServiceFilter` is provided for managing timeouts by services.

Result:
HTTP API provides request/response transaction timeout capabilities

servicetalk-concurrent-api/src/test/java/io/servicetalk/concurrent/api/completable/TimeoutCompletableTest.java

@@ -30,6 +30,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import javax.annotation.Nullable;
@@ -135,6 +136,25 @@ public void justSubscribeTimeout() {
         assertThat(listener.awaitOnError(), instanceOf(TimeoutException.class));
     }
 
+    @Test
+    public void cancelDoesOnError() throws Exception {
+        DelayedOnSubscribeCompletable delayedCompletable = new DelayedOnSubscribeCompletable();
+        init(delayedCompletable, false);
+        CountDownLatch cancelLatch = new CountDownLatch(1);
+        CompletableSource.Subscriber subscriber = delayedCompletable.subscriber;
+        assertNotNull(subscriber);
+        subscriber.onSubscribe(() -> {
+            subscriber.onError(DELIBERATE_EXCEPTION);
+            cancelLatch.countDown();
+        });
+        testExecutor.advanceTimeBy(1, NANOSECONDS);
+        assertThat(testExecutor.scheduledTasksPending(), is(0));
+        assertThat(testExecutor.scheduledTasksExecuted(), is(1));
+        cancelLatch.await();
+        Throwable error = listener.awaitOnError();
+        assertThat(error, instanceOf(TimeoutException.class));
+    }
+
     private void init() {
         init(source.ignoreElement(), true);
     }

servicetalk-examples/README.adoc

@@ -10,6 +10,7 @@ as shown here:
 ----
 implementation project(":servicetalk-annotations")
 implementation project(":servicetalk-http-netty")
+implementation project(":servicetalk-http-utils")
 ----
 
 In actual user projects, ServiceTalk modules would be referenced via standard artifacts coordinates,
@@ -23,4 +24,5 @@ implementation platform("io.servicetalk:servicetalk-bom:$serviceTalkVersion")
 // The version for all ServiceTalk dependencies will be resolved based on information in `servicetalk-bom`.
 implementation "io.servicetalk:servicetalk-annotations"
 implementation "io.servicetalk:servicetalk-http-netty"
+implementation "io.servicetalk:servicetalk-http-utils"
 ----

servicetalk-examples/docs/modules/ROOT/pages/_partials/nav-versioned.adoc

@@ -3,6 +3,7 @@
 ** xref:{page-version}@servicetalk-examples::http/index.adoc#HelloWorld[Hello World]
 ** xref:{page-version}@servicetalk-examples::http/index.adoc#Compression[Compression]
 ** xref:{page-version}@servicetalk-examples::http/index.adoc#Debugging[Debugging]
+** xref:{page-version}@servicetalk-examples::http/index.adoc#Timeout[Timeout]
 ** xref:{page-version}@servicetalk-examples::http/index.adoc#Serialization[Serialization]
 ** xref:{page-version}@servicetalk-examples::http/index.adoc#JAXRS[JAX-RS]
 ** xref:{page-version}@servicetalk-examples::http/index.adoc#MetaData[MetaData]

servicetalk-examples/docs/modules/ROOT/pages/http/index.adoc

@@ -92,6 +92,16 @@ Extends the async "Hello World" example to demonstrate some useful features avai
  server enhanced with debugging capabilities.
 * link:{source-root}/servicetalk-examples/http/debugging/src/main/java/io/servicetalk/examples/http/debugging/DebuggingExampleClient.java[DebuggingExampleClient.java] - the async `Hello World!` client enhanced with debugging capabilities.
 
+[#Timeout]
+== Timeout
+
+Extends the async "Hello World" example to demonstrate the use of timeout filters and operators. You should read and
+ understand the async "Hello World" example first to understand the additions this example adds. No separate example is
+ needed for the other API variants as the usage of the debugging features are the same for all API styles.
+
+* link:{source-root}/servicetalk-examples/http/timeout/src/main/java/io/servicetalk/examples/http/timeout/TimeoutServer.java[TimeoutServer] - the async `Hello World!` server client enhanced to use timeout capabilities.
+* link:{source-root}/servicetalk-examples/http/timeout/src/main/java/io/servicetalk/examples/http/timeout/DebuggingExampleClient.java[TimeoutClient.java] - the async `Hello World!` client enhanced to use timeout capabilities.
+
 [#Serialization]
 == Serialization
 

servicetalk-examples/http/helloworld/src/main/java/io/servicetalk/examples/http/helloworld/async/HelloWorldClient.java

@@ -37,6 +37,7 @@ public static void main(String[] args) throws Exception {
                         System.out.println(resp.payloadBody(textDeserializer()));
                     });
 
+            // block until request is complete and afterFinally() is called
             responseProcessedLatch.await();
         }
     }

servicetalk-examples/http/timeout/build.gradle

@@ -0,0 +1,25 @@
+/*
+ * Copyright © 2019, 2021 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed 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.
+ */
+
+apply plugin: "java"
+apply from: "../../gradle/idea.gradle"
+
+dependencies {
+  implementation project(":servicetalk-http-netty")
+  implementation project(":servicetalk-http-utils")
+
+  runtimeOnly "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
+}

servicetalk-examples/http/timeout/src/main/java/io/servicetalk/examples/http/timeout/TimeoutClient.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright © 2021 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed 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 io.servicetalk.examples.http.timeout;
+
+import io.servicetalk.http.api.HttpClient;
+import io.servicetalk.http.api.SingleAddressHttpClientBuilder;
+import io.servicetalk.http.netty.HttpClients;
+import io.servicetalk.http.utils.TimeoutHttpRequesterFilter;
+import io.servicetalk.transport.api.HostAndPort;
+
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+
+import static io.servicetalk.http.api.HttpSerializationProviders.textDeserializer;
+
+/**
+ * Extends the async 'Hello World!' example to demonstrate use of timeout filters and timeout operators. If a single
+ * timeout can be applied to all transactions then the timeout should be applied using the
+ * {@link TimeoutHttpRequesterFilter}. If only some transactions require a timeout then the timeout should be applied
+ * using a {@link io.servicetalk.concurrent.api.Single#timeout(Duration)} Single.timeout()} or a
+ * {@link io.servicetalk.concurrent.api.Publisher#timeoutTerminal(Duration)} (Duration)} Publisher.timeoutTerminal()}
+ * operator.
+ */
+public final class TimeoutClient {
+
+    public static void main(String[] args) throws Exception {
+        SingleAddressHttpClientBuilder<HostAndPort, InetSocketAddress> builder =
+                HttpClients.forSingleAddress("localhost", 8080)
+                        // Filter enforces that requests made with this client must fully complete
+                        // within 10 seconds or will be cancelled.
+                        .appendClientFilter(new TimeoutHttpRequesterFilter(Duration.ofSeconds(10), true));
+
+        try (HttpClient client = builder.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(2);
+
+            // first request, with default timeout from HttpClient (this will succeed)
+            client.request(client.get("/sayHello"))
+                    .afterFinally(responseProcessedLatch::countDown)
+                    .afterOnError(System.err::println)
+                    .subscribe(resp -> {
+                        System.out.println(resp.toString((name, value) -> value));
+                        System.out.println(resp.payloadBody(textDeserializer()));
+                    });
+
+            // second request, with custom timeout that is lower than the client default (this will timeout)
+            client.request(client.get("/sayHello"))
+                    // This request and response must complete within 3 seconds or the request will be cancelled.
+                    .timeout(Duration.ofSeconds(3))
+                    .afterFinally(responseProcessedLatch::countDown)
+                    .afterOnError(System.err::println)
+                    .subscribe(resp -> {
+                        System.out.println(resp.toString((name, value) -> value));
+                        System.out.println(resp.payloadBody(textDeserializer()));
+                    });
+
+            // block until requests are complete and afterFinally() has been called
+            responseProcessedLatch.await();
+        }
+    }
+}

servicetalk-examples/http/timeout/src/main/java/io/servicetalk/examples/http/timeout/TimeoutServer.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright © 2021 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed 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 io.servicetalk.examples.http.timeout;
+
+import io.servicetalk.concurrent.api.Single;
+import io.servicetalk.http.netty.HttpServers;
+import io.servicetalk.http.utils.TimeoutHttpServiceFilter;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import static io.servicetalk.concurrent.api.Single.succeeded;
+import static io.servicetalk.http.api.HttpSerializationProviders.textSerializer;
+
+/**
+ * Extends the async 'Hello World!' example to demonstrate use of timeout filter.
+ */
+public final class TimeoutServer {
+    public static void main(String[] args) throws Exception {
+        HttpServers.forPort(8080)
+                // Filter enforces that responses must complete within 30 seconds or will be cancelled.
+                .appendServiceFilter(new TimeoutHttpServiceFilter(Duration.ofSeconds(30)))
+                .listenAndAwait((ctx, request, responseFactory) ->
+                        Single.defer(() -> {
+                            // Force a 5 second delay in the response.
+                            try {
+                                TimeUnit.SECONDS.sleep(5);
+                            } catch (InterruptedException woken) {
+                                Thread.interrupted();
+                                // just continue
+                            }
+
+                            return succeeded(responseFactory.ok()
+                                    .payloadBody("Hello World!", textSerializer()))
+                                    .subscribeShareContext();
+                        }))
+                .awaitShutdown();
+    }
+}

servicetalk-examples/http/timeout/src/main/resources/log4j2.xml

@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright © 2018 Apple Inc. and the ServiceTalk project authors
+  ~
+  ~ Licensed 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.
+  -->
+<Configuration status="info">
+  <Appenders>
+    <Console name="Console" target="SYSTEM_OUT">
+      <PatternLayout pattern="%d %30t [%-5level] %-30logger{1} - %msg%n"/>
+    </Console>
+  </Appenders>
+  <Loggers>
+    <!-- Prints server start and shutdown -->
+    <Logger name="io.servicetalk.http.netty.NettyHttpServer" level="DEBUG"/>
+
+    <!-- Prints default subscriber errors-->
+    <Logger name="io.servicetalk.concurrent.api" level="DEBUG"/>
+
+    <!-- Use `-Dservicetalk.logger.level=DEBUG` to change the root logger level via command line  -->
+    <Root level="${sys:servicetalk.logger.level:-INFO}">
+      <AppenderRef ref="Console"/>
+    </Root>
+  </Loggers>
+</Configuration>