Reaper is not a Future

Author: Stephane Landelle

src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

@@ -566,10 +566,10 @@ public void operationComplete(ChannelFuture cf) {
             int requestTimeout = requestTimeoutInMs(config, future.getRequest().getPerRequestConfig());
             int schedulePeriod = requestTimeout != -1 ? (config.getIdleConnectionTimeoutInMs() != -1 ? Math.min(requestTimeout, config.getIdleConnectionTimeoutInMs()) : requestTimeout) : config.getIdleConnectionTimeoutInMs();
             if (schedulePeriod != -1 && !future.isDone() && !future.isCancelled()) {
-                ReaperFuture reaperFuture = new ReaperFuture(future);
-                Future<?> scheduledFuture = config.reaper().scheduleAtFixedRate(reaperFuture, 0, schedulePeriod, TimeUnit.MILLISECONDS);
-                reaperFuture.setScheduledFuture(scheduledFuture);
-                future.setReaperFuture(reaperFuture);
+                Reaper reaper = new Reaper(future);
+                Future<?> scheduledFuture = config.reaper().scheduleAtFixedRate(reaper, 0, schedulePeriod, TimeUnit.MILLISECONDS);
+                reaper.setScheduledFuture(scheduledFuture);
+                future.setReaper(reaper);
             }
         } catch (RejectedExecutionException ex) {
             abort(future, ex);
@@ -866,7 +866,7 @@ public void close() {
                 ChannelHandlerContext ctx = channel.getPipeline().getContext(NettyAsyncHttpProvider.class);
                 if (ctx.getAttachment() instanceof NettyResponseFuture<?>) {
                     NettyResponseFuture<?> future = (NettyResponseFuture<?>) ctx.getAttachment();
-                    future.setReaperFuture(null);
+                    future.setReaper(null);
                 }
             }
 
@@ -1726,54 +1726,23 @@ public void operationProgressed(ChannelFuture cf, long amount, long current, lon
      * Because some implementation of the ThreadSchedulingService do not clean up cancel task until they try to run them, we wrap the task with the future so the when the NettyResponseFuture cancel the reaper future this wrapper will release the references to the channel and the
      * nettyResponseFuture immediately. Otherwise, the memory referenced this way will only be released after the request timeout period which can be arbitrary long.
      */
-    private final class ReaperFuture implements Future, Runnable {
+    public final class Reaper implements Runnable {
         private Future scheduledFuture;
         private NettyResponseFuture<?> nettyResponseFuture;
 
-        public ReaperFuture(NettyResponseFuture<?> nettyResponseFuture) {
+        public Reaper(NettyResponseFuture<?> nettyResponseFuture) {
             this.nettyResponseFuture = nettyResponseFuture;
         }
 
         public void setScheduledFuture(Future scheduledFuture) {
             this.scheduledFuture = scheduledFuture;
         }
 
-        /**
-         * @Override
-         */
         public boolean cancel(boolean mayInterruptIfRunning) {
             nettyResponseFuture = null;
             return scheduledFuture.cancel(mayInterruptIfRunning);
         }
 
-        /**
-         * @Override
-         */
-        public Object get() throws InterruptedException, ExecutionException {
-            return scheduledFuture.get();
-        }
-
-        /**
-         * @Override
-         */
-        public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
-            return scheduledFuture.get(timeout, unit);
-        }
-
-        /**
-         * @Override
-         */
-        public boolean isCancelled() {
-            return scheduledFuture.isCancelled();
-        }
-
-        /**
-         * @Override
-         */
-        public boolean isDone() {
-            return scheduledFuture.isDone();
-        }
-
         private void expire(String message) {
             log.debug("{} for {}", message, nettyResponseFuture);
             abort(nettyResponseFuture, new TimeoutException(message));
@@ -1800,6 +1769,7 @@ public synchronized void run() {
                 if (nettyResponseFuture.hasRequestTimedOut(now))
                     message = "Request reached time out of " + nettyResponseFuture.getRequestTimeoutInMs() + " ms after " + age + " ms";
                 else
+                    // If the Reaper wakes up, there's only 2 possibilities: request timeout or idle
                     message = "Request reached idle time out of " + nettyResponseFuture.getIdleConnectionTimeoutInMs() + " ms after " + age + " ms";
                 expire(message);
 

src/main/java/com/ning/http/client/providers/netty/NettyResponseFuture.java

@@ -42,6 +42,7 @@
 import com.ning.http.client.ProxyServer;
 import com.ning.http.client.Request;
 import com.ning.http.client.listenable.AbstractListenableFuture;
+import com.ning.http.client.providers.netty.NettyAsyncHttpProvider.Reaper;
 
 /**
  * A {@link Future} that can be used to track when an asynchronous HTTP request has been fully processed.
@@ -71,7 +72,7 @@ enum STATE {
     private HttpResponse httpResponse;
     private final AtomicReference<ExecutionException> exEx = new AtomicReference<ExecutionException>();
     private final AtomicInteger redirectCount = new AtomicInteger();
-    private volatile Future<?> reaperFuture;
+    private volatile Reaper reaper;
     private final AtomicBoolean inAuth = new AtomicBoolean(false);
     private final AtomicBoolean statusReceived = new AtomicBoolean(false);
     private final AtomicLong touch = new AtomicLong(millisTime());
@@ -216,8 +217,8 @@ public V get() throws InterruptedException, ExecutionException {
     }
 
     void cancelReaper() {
-        if (reaperFuture != null) {
-            reaperFuture.cancel(true);
+        if (reaper != null) {
+            reaper.cancel(true);
         }
     }
 
@@ -387,9 +388,9 @@ protected int incrementAndGetCurrentRedirectCount() {
         return redirectCount.incrementAndGet();
     }
 
-    protected void setReaperFuture(Future<?> reaperFuture) {
+    protected void setReaper(Reaper reaper) {
         cancelReaper();
-        this.reaperFuture = reaperFuture;
+        this.reaper = reaper;
     }
 
     protected boolean isInAuth() {
@@ -520,7 +521,7 @@ public String toString() {
                 ",\n\thttpResponse=" + httpResponse + //
                 ",\n\texEx=" + exEx + //
                 ",\n\tredirectCount=" + redirectCount + //
-                ",\n\treaperFuture=" + reaperFuture + //
+                ",\n\treaper=" + reaper + //
                 ",\n\tinAuth=" + inAuth + //
                 ",\n\tstatusReceived=" + statusReceived + //
                 ",\n\ttouch=" + touch + //