Fix Listener exec when it's added to the ListenableFuture after completion, close AsyncHttpClient#1128

Author: slandelle

client/src/main/java/org/asynchttpclient/future/AbstractListenableFuture.java

@@ -45,6 +45,7 @@ public abstract class AbstractListenableFuture<V> implements ListenableFuture<V>
 
     // The execution list to hold our executors.
     // lazy fields
+    private volatile boolean hasRun;
     private volatile boolean executionListInitialized;
     private ExecutionList executionList;
 
@@ -62,19 +63,20 @@ private ExecutionList executionList() {
         return executionListInitialized ? executionList : lazyExecutionList();
     }
 
-    /*
-     * Adds a listener/executor pair to execution list to execute when this task is completed.
-     */
-
+    @Override
     public ListenableFuture<V> addListener(Runnable listener, Executor exec) {
         executionList().add(listener, exec);
+        if (hasRun) {
+            runListeners();
+        }
         return this;
     }
 
-    /*
+    /**
      * Execute the execution list.
      */
     protected void runListeners() {
+        hasRun = true;
         if (executionListInitialized) {
             executionList().run();
         }

client/src/test/java/org/asynchttpclient/ListenableFutureTest.java

@@ -12,7 +12,7 @@
  */
 package org.asynchttpclient;
 
-import static org.asynchttpclient.Dsl.*;
+import static org.asynchttpclient.Dsl.asyncHttpClient;
 import static org.testng.Assert.assertEquals;
 
 import java.util.concurrent.CountDownLatch;
@@ -49,4 +49,33 @@ public void run() {
             assertEquals(statusCode.get(), 200);
         }
     }
+
+    @Test(groups = "standalone")
+    public void testListenableFutureAfterCompletion() throws Exception {
+
+        AtomicInteger counter = new AtomicInteger(1);
+
+        try (AsyncHttpClient ahc = asyncHttpClient()) {
+            final ListenableFuture<Response> future = ahc.prepareGet(getTargetUrl()).execute();
+            future.get();
+            future.addListener(() -> counter.decrementAndGet(), Runnable::run);
+        }
+        assertEquals(0, counter.get());
+    }
+
+    @Test(groups = "standalone")
+    public void testListenableFutureBeforeAndAfterCompletion() throws Exception {
+
+        AtomicInteger counter = new AtomicInteger(2);
+
+        try (AsyncHttpClient ahc = asyncHttpClient()) {
+            final ListenableFuture<Response> future = ahc.prepareGet(getTargetUrl()).execute();
+
+            future.addListener(() -> counter.decrementAndGet(), Runnable::run);
+
+            future.get();
+            future.addListener(() -> counter.decrementAndGet(), Runnable::run);
+        }
+        assertEquals(0, counter.get());
+    }
 }