Discard new writes without closing connection on AbortWritesEvent (#…

…3102)

Motivation:

Behavior of `SslHandler` in Netty 4.1.115 changed. Now it can wrap data and flush them to the network even if flush was not requested. In result, it affected our test `ConnectionCloseHeaderHandlingTest.PipelinedRequestsTest.serverCloseSecondPipelinedRequestWriteAborted` that we had to disable in #3097 to proceed with the upgrade.
Debugging showed that when we receive `AbortWritesEvent`, the following behavior of `WriteStreamSubscriber` depends on how many `activeWrites` we have. With 4.1.115 netty will flush data, there will be 0 `activeWrites` and it will close (with reset) the outbound. In result, we lose data of the previous response in the pipeline that we haven't read yet.

Modifications:
- Use `listenerDiscard` instead of `channelClosed` that will make sure new writes will be discarded without prematurely affecting the connection state regardless of the `activeWrites` count.
- Unskip the test.

Result:

Graceful handling of `connection: close` header does not depend on flushing behavior.

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

@@ -390,8 +390,6 @@ void serverCloseTwoPipelinedRequestsSentBeforeFirstResponse(boolean useUds, bool
         @MethodSource("io.servicetalk.http.netty.ConnectionCloseHeaderHandlingTest#pipelinedRequestsTestData")
         void serverCloseSecondPipelinedRequestWriteAborted(boolean useUds, boolean viaProxy,
                                                            boolean awaitRequestPayload) throws Exception {
-            assumeFalse(viaProxy, "Let all other tests run with Netty 4.1.115");
-
             setUp(useUds, viaProxy, awaitRequestPayload);
             AtomicReference<Throwable> secondRequestError = new AtomicReference<>();
             CountDownLatch secondResponseReceived = new CountDownLatch(1);

servicetalk-transport-netty-internal/src/main/java/io/servicetalk/transport/netty/internal/DefaultNettyConnection.java

@@ -937,7 +937,7 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
             } else if (evt == CloseHandler.OutboundDataEndEvent.INSTANCE) {
                 connection.channelOutboundListener.channelOutboundClosed();
             } else if (evt == AbortWritesEvent.INSTANCE) {
-                connection.channelOutboundListener.channelClosed(StacklessClosedChannelException.newInstance(
+                connection.channelOutboundListener.listenerDiscard(StacklessClosedChannelException.newInstance(
                         DefaultNettyConnection.class, "userEventTriggered(AbortWritesEvent)"));
             } else if (evt == ChannelOutputShutdownEvent.INSTANCE) {
                 connection.closeHandler.channelClosedOutbound(ctx);