Improve exceptional behavior in reactive streams

* Errors on the request are now propagated to reactive subscribers
  instead of just to the request's ListenableFuture
* Read timeouts can no longer occur if a reactive streams subscriber has
  no outstanding request. Note that this does not affect request
  timeouts - only read timeouts.

Author: n-miles

Diff for: client/src/main/java/org/asynchttpclient/netty/handler/HttpHandler.java

@@ -39,8 +39,7 @@ public HttpHandler(AsyncHttpClientConfig config, ChannelManager channelManager,
     super(config, channelManager, requestSender);
   }
 
-  private boolean abortAfterHandlingStatus(//
-                                           AsyncHandler<?> handler,
+  private boolean abortAfterHandlingStatus(AsyncHandler<?> handler,
                                            NettyResponseStatus status) throws Exception {
     return handler.onStatusReceived(status) == State.ABORT;
   }

Diff for: client/src/main/java/org/asynchttpclient/netty/handler/StreamedResponsePublisher.java

@@ -14,10 +14,15 @@
 
 import com.typesafe.netty.HandlerPublisher;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.concurrent.EventExecutor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import org.asynchttpclient.HttpResponseBodyPart;
 import org.asynchttpclient.netty.NettyResponseFuture;
 import org.asynchttpclient.netty.channel.ChannelManager;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -28,6 +33,8 @@ public class StreamedResponsePublisher extends HandlerPublisher<HttpResponseBody
   private final ChannelManager channelManager;
   private final NettyResponseFuture<?> future;
   private final Channel channel;
+  private final AtomicBoolean hasOutstandingRequest = new AtomicBoolean(false);
+  private Throwable error;
 
   StreamedResponsePublisher(EventExecutor executor, ChannelManager channelManager, NettyResponseFuture<?> future, Channel channel) {
     super(executor, HttpResponseBodyPart.class);
@@ -51,7 +58,66 @@ protected void cancelled() {
     channelManager.closeChannel(channel);
   }
 
+  @Override
+  protected void requestDemand() {
+    hasOutstandingRequest.set(true);
+    super.requestDemand();
+  }
+
+  @Override
+  public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+    hasOutstandingRequest.set(false);
+    super.channelReadComplete(ctx);
+  }
+
+  @Override
+  public void subscribe(Subscriber<? super HttpResponseBodyPart> subscriber) {
+    super.subscribe(new ErrorReplacingSubscriber(subscriber));
+  }
+
+  public boolean hasOutstandingRequest() {
+    return hasOutstandingRequest.get();
+  }
+
   NettyResponseFuture<?> future() {
     return future;
   }
+
+  public void setError(Throwable t) {
+    this.error = t;
+  }
+
+  private class ErrorReplacingSubscriber implements Subscriber<HttpResponseBodyPart> {
+
+    Subscriber<? super HttpResponseBodyPart> subscriber;
+
+    ErrorReplacingSubscriber(Subscriber<? super HttpResponseBodyPart> subscriber) {
+      this.subscriber = subscriber;
+    }
+
+    @Override
+    public void onSubscribe(Subscription s) {
+      subscriber.onSubscribe(s);
+    }
+
+    @Override
+    public void onNext(HttpResponseBodyPart httpResponseBodyPart) {
+      subscriber.onNext(httpResponseBodyPart);
+    }
+
+    @Override
+    public void onError(Throwable t) {
+      subscriber.onError(t);
+    }
+
+    @Override
+    public void onComplete() {
+      Throwable replacementError = error;
+      if (replacementError == null) {
+        subscriber.onComplete();
+      } else {
+        subscriber.onError(replacementError);
+      }
+    }
+  }
 }

Diff for: client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -35,6 +35,7 @@
 import org.asynchttpclient.netty.OnLastHttpContentCallback;
 import org.asynchttpclient.netty.SimpleFutureListener;
 import org.asynchttpclient.netty.channel.*;
+import org.asynchttpclient.netty.handler.StreamedResponsePublisher;
 import org.asynchttpclient.netty.timeout.TimeoutsHolder;
 import org.asynchttpclient.proxy.ProxyServer;
 import org.asynchttpclient.resolver.RequestHostnameResolver;
@@ -462,8 +463,15 @@ private void scheduleReadTimeout(NettyResponseFuture<?> nettyResponseFuture) {
 
   public void abort(Channel channel, NettyResponseFuture<?> future, Throwable t) {
 
-    if (channel != null && channel.isActive()) {
-      channelManager.closeChannel(channel);
+    if (channel != null) {
+      Object attribute = Channels.getAttribute(future.channel());
+      if (attribute instanceof StreamedResponsePublisher) {
+        ((StreamedResponsePublisher) attribute).setError(t);
+      }
+
+      if (channel.isActive()) {
+        channelManager.closeChannel(channel);
+      }
     }
 
     if (!future.isDone()) {

Diff for: client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java

@@ -15,6 +15,8 @@
 
 import io.netty.util.Timeout;
 import org.asynchttpclient.netty.NettyResponseFuture;
+import org.asynchttpclient.netty.channel.Channels;
+import org.asynchttpclient.netty.handler.StreamedResponsePublisher;
 import org.asynchttpclient.netty.request.NettyRequestSender;
 import org.asynchttpclient.util.StringBuilderPool;
 
@@ -47,7 +49,7 @@ public void run(Timeout timeout) {
     long currentReadTimeoutInstant = readTimeout + nettyResponseFuture.getLastTouch();
     long durationBeforeCurrentReadTimeout = currentReadTimeoutInstant - now;
 
-    if (durationBeforeCurrentReadTimeout <= 0L) {
+    if (durationBeforeCurrentReadTimeout <= 0L && !isReactiveWithNoOutstandingRequest()) {
       // idleConnectTimeout reached
       StringBuilder sb = StringBuilderPool.DEFAULT.stringBuilder().append("Read timeout to ");
       appendRemoteAddress(sb);
@@ -62,4 +64,10 @@ public void run(Timeout timeout) {
       timeoutsHolder.startReadTimeout(this);
     }
   }
+
+  private boolean isReactiveWithNoOutstandingRequest() {
+    Object attribute = Channels.getAttribute(nettyResponseFuture.channel());
+    return attribute instanceof StreamedResponsePublisher &&
+            !((StreamedResponsePublisher) attribute).hasOutstandingRequest();
+  }
 }

Diff for: client/src/test/java/org/asynchttpclient/reactivestreams/ReactiveStreamsDownLoadTest.java renamed to client/src/test/java/org/asynchttpclient/reactivestreams/ReactiveStreamsDownloadTest.java

@@ -39,11 +39,11 @@
 import static org.asynchttpclient.Dsl.asyncHttpClient;
 import static org.testng.Assert.assertEquals;
 
-public class ReactiveStreamsDownLoadTest {
+public class ReactiveStreamsDownloadTest {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(ReactiveStreamsDownLoadTest.class);
+  private static final Logger LOGGER = LoggerFactory.getLogger(ReactiveStreamsDownloadTest.class);
 
-  private int serverPort = 8080;
+  private final int serverPort = 8080;
   private File largeFile;
   private File smallFile;
 
@@ -104,7 +104,7 @@ public void onThrowable(Throwable t) {
     }
 
     @Override
-    public State onBodyPartReceived(HttpResponseBodyPart bodyPart) throws Exception {
+    public State onBodyPartReceived(HttpResponseBodyPart bodyPart) {
       LOGGER.debug("SimpleStreamedAsyncHandleronCompleted onBodyPartReceived");
       throw new AssertionError("Should not have received body part");
     }
@@ -115,12 +115,12 @@ public State onStatusReceived(HttpResponseStatus responseStatus) {
     }
 
     @Override
-    public State onHeadersReceived(HttpHeaders headers) throws Exception {
+    public State onHeadersReceived(HttpHeaders headers) {
       return State.CONTINUE;
     }
 
     @Override
-    public SimpleStreamedAsyncHandler onCompleted() throws Exception {
+    public SimpleStreamedAsyncHandler onCompleted() {
       LOGGER.debug("SimpleStreamedAsyncHandleronCompleted onSubscribe");
       return this;
     }