Ensure clients are finally closed, fix AsyncHttpClient#239

Author: Stephane Landelle

src/test/java/com/ning/http/client/async/AsyncProvidersBasicTest.java

@@ -47,7 +47,7 @@
 
 /**
  * Tests default asynchronous life cycle.
- *
+ * 
  * @author Hubert Iwaniuk
  */
 public abstract class AsyncStreamLifecycleTest extends AbstractBasicTest {
@@ -63,8 +63,7 @@ public void tearDownGlobal() throws Exception {
     @Override
     public AbstractHandler configureHandler() throws Exception {
         return new AbstractHandler() {
-            public void handle(String s, Request request, HttpServletRequest req, final HttpServletResponse resp)
-                    throws IOException, ServletException {
+            public void handle(String s, Request request, HttpServletRequest req, final HttpServletResponse resp) throws IOException, ServletException {
                 resp.setContentType("text/plain;charset=utf-8");
                 resp.setStatus(200);
                 final Continuation continuation = ContinuationSupport.getContinuation(req);
@@ -100,62 +99,64 @@ public void run() {
         };
     }
 
-    //TODO Netty only.
+    // TODO Netty only.
 
-    @Test(groups = {"standalone", "default_provider"})
+    @Test(groups = { "standalone", "default_provider" })
     public void testStream() throws IOException {
-        AsyncHttpClient ahc = getAsyncHttpClient(null);
-        final AtomicBoolean err = new AtomicBoolean(false);
-        final LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<String>();
-        final AtomicBoolean status = new AtomicBoolean(false);
-        final AtomicInteger headers = new AtomicInteger(0);
-        final CountDownLatch latch = new CountDownLatch(1);
-        ahc.executeRequest(ahc.prepareGet(getTargetUrl()).build(), new AsyncHandler<Object>() {
-            public void onThrowable(Throwable t) {
-                fail("Got throwable.", t);
-                err.set(true);
-            }
+        AsyncHttpClient client = getAsyncHttpClient(null);
+        try {
+            final AtomicBoolean err = new AtomicBoolean(false);
+            final LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<String>();
+            final AtomicBoolean status = new AtomicBoolean(false);
+            final AtomicInteger headers = new AtomicInteger(0);
+            final CountDownLatch latch = new CountDownLatch(1);
+            client.executeRequest(client.prepareGet(getTargetUrl()).build(), new AsyncHandler<Object>() {
+                public void onThrowable(Throwable t) {
+                    fail("Got throwable.", t);
+                    err.set(true);
+                }
 
-            public STATE onBodyPartReceived(HttpResponseBodyPart e) throws Exception {
-                String s = new String(e.getBodyPartBytes());
-                log.info("got part: {}", s);
-                if (s.isEmpty()) {
-                    //noinspection ThrowableInstanceNeverThrown
-                    log.warn("Sampling stacktrace.",
-                            new Throwable("trace that, we should not get called for empty body."));
+                public STATE onBodyPartReceived(HttpResponseBodyPart e) throws Exception {
+                    String s = new String(e.getBodyPartBytes());
+                    log.info("got part: {}", s);
+                    if (s.isEmpty()) {
+                        // noinspection ThrowableInstanceNeverThrown
+                        log.warn("Sampling stacktrace.", new Throwable("trace that, we should not get called for empty body."));
+                    }
+                    queue.put(s);
+                    return STATE.CONTINUE;
                 }
-                queue.put(s);
-                return STATE.CONTINUE;
-            }
 
-            public STATE onStatusReceived(HttpResponseStatus e) throws Exception {
-                status.set(true);
-                return STATE.CONTINUE;
-            }
+                public STATE onStatusReceived(HttpResponseStatus e) throws Exception {
+                    status.set(true);
+                    return STATE.CONTINUE;
+                }
 
-            public STATE onHeadersReceived(HttpResponseHeaders e) throws Exception {
-                if (headers.incrementAndGet() == 2) {
-                    throw new Exception("Analyze this.");
+                public STATE onHeadersReceived(HttpResponseHeaders e) throws Exception {
+                    if (headers.incrementAndGet() == 2) {
+                        throw new Exception("Analyze this.");
+                    }
+                    return STATE.CONTINUE;
                 }
-                return STATE.CONTINUE;
-            }
 
-            public Object onCompleted() throws Exception {
-                latch.countDown();
-                return null;
+                public Object onCompleted() throws Exception {
+                    latch.countDown();
+                    return null;
+                }
+            });
+            try {
+                assertTrue(latch.await(1, TimeUnit.SECONDS), "Latch failed.");
+            } catch (InterruptedException e) {
+                fail("Interrupted.", e);
             }
-        });
-        try {
-            assertTrue(latch.await(1, TimeUnit.SECONDS), "Latch failed.");
-        } catch (InterruptedException e) {
-            fail("Interrupted.", e);
+            assertFalse(err.get());
+            assertEquals(queue.size(), 2);
+            assertTrue(queue.contains("part1"));
+            assertTrue(queue.contains("part2"));
+            assertTrue(status.get());
+            assertEquals(headers.get(), 1);
+        } finally {
+            client.close();
         }
-        assertFalse(err.get());
-        assertEquals(queue.size(), 2);
-        assertTrue(queue.contains("part1"));
-        assertTrue(queue.contains("part2"));
-        assertTrue(status.get());
-        assertEquals(headers.get(), 1);
-        ahc.close();
     }
 }