make control flow in ConnectionCloseHeaderHandlingTest and

GracefulConnectionClosureHandlingTest more robust

Author: Scottmitch

servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/ConnectionCloseHeaderHandlingTest.java

@@ -24,6 +24,7 @@
 import io.servicetalk.http.api.StreamingHttpClient;
 import io.servicetalk.http.api.StreamingHttpRequest;
 import io.servicetalk.http.api.StreamingHttpResponse;
+import io.servicetalk.oio.api.internal.PayloadWriterUtils;
 import io.servicetalk.test.resources.DefaultTestCerts;
 import io.servicetalk.transport.api.ClientSslConfigBuilder;
 import io.servicetalk.transport.api.ConnectionContext;
@@ -168,16 +169,17 @@ public Completable accept(final ConnectionContext context) {
                             // Subscribe to the request payload body before response writer closes
                             BlockingIterator<Buffer> iterator = request.payloadBody().iterator();
                             // Consume request payload body asynchronously:
-                            ctx.executionContext().executor().execute(() -> {
+                            Future<Void> writeFuture = ctx.executionContext().executor().submit(() -> {
                                 while (iterator.hasNext()) {
                                     Buffer chunk = iterator.next();
                                     assert chunk != null;
                                     requestPayloadSize.addAndGet(chunk.readableBytes());
                                 }
-                                requestPayloadReceived.countDown();
-                            });
+                            }).beforeOnError(cause -> PayloadWriterUtils.safeClose(writer, cause))
+                            .afterFinally(requestPayloadReceived::countDown)
+                            .toFuture();
                             if (awaitRequestPayload) {
-                                requestPayloadReceived.await();
+                                writeFuture.get();
                             }
                             if (!noResponseContent) {
                                 // Defer payload body to see how client-side processes "Connection: close" header

servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/GracefulConnectionClosureHandlingTest.java

@@ -29,6 +29,7 @@
 import io.servicetalk.http.api.StreamingHttpClient;
 import io.servicetalk.http.api.StreamingHttpRequest;
 import io.servicetalk.http.api.StreamingHttpResponse;
+import io.servicetalk.oio.api.internal.PayloadWriterUtils;
 import io.servicetalk.test.resources.DefaultTestCerts;
 import io.servicetalk.transport.api.ClientSslConfigBuilder;
 import io.servicetalk.transport.api.ConnectionContext;
@@ -189,23 +190,18 @@ public Completable accept(final ConnectionContext context) {
                 // Subscribe to the request payload body before response writer closes
                 BlockingIterator<Buffer> iterator = request.payloadBody().iterator();
                 // Consume request payload body asynchronously:
-                ctx.executionContext().executor().execute(() -> {
+                ctx.executionContext().executor().submit(() -> {
                     int receivedSize = 0;
                     while (iterator.hasNext()) {
-                        try {
-                            Buffer chunk = iterator.next();
-                            assert chunk != null;
-                            receivedSize += chunk.readableBytes();
-                        } catch (Throwable cause) {
-                            try {
-                                writer.write(cause.toString());
-                            } catch (IOException e) {
-                                throw new AssertionError(e);
-                            }
-                        }
+                        Buffer chunk = iterator.next();
+                        assert chunk != null;
+                        receivedSize += chunk.readableBytes();
                     }
                     serverReceivedRequestPayload.add(receivedSize);
-                });
+                }).beforeOnError(cause -> {
+                    serverReceivedRequestPayload.add(-1);
+                    PayloadWriterUtils.safeClose(writer, cause);
+                }).toFuture();
                 serverSendResponsePayload.await();
                 writer.write(RESPONSE_CONTENT);
             }