Netty concurrent.ExecutionException about async-http-client

[survivant]

here my use case (client side.)  (same usecase for issue 107 108 109)

@Test(groups = {"standalone", "default_provider"})
    public void loginDuplicateUsernameXHRPollingTest() throws Throwable {
        final AsyncHttpClient client = getAsyncHttpClient(new AsyncHttpClientConfig.Builder().setFollowRedirects(true).build());

        final String sessionid1 = getSessionID(client, GET_SESSION_URL);

        final String username = "test_" + System.currentTimeMillis();

        // maintenant on fait login
        login("clientXHRPolling1", client, GET_SESSION_URL+"xhr-polling/" + sessionid1, username, true);

        final AsyncHttpClient client2 = getAsyncHttpClient(new AsyncHttpClientConfig.Builder().setFollowRedirects(true).build());

        final String sessionid2 = getSessionID(client, GET_SESSION_URL);

        // maintenant on fait login
        login("clientXHRPolling2", client2, GET_SESSION_URL+"xhr-polling/" + sessionid2, username, false);

        client.close();
        client2.close();
    }


public static AsyncHttpClient getAsyncHttpClient(AsyncHttpClientConfig config) {
        if (config == null) {
            config = new AsyncHttpClientConfig.Builder().build();
        }
        //return new AsyncHttpClient(new GrizzlyAsyncHttpProvider(config), config);
        return new AsyncHttpClient(new NettyAsyncHttpProvider(config), config);
    }

public static String getSessionID(AsyncHttpClient client, String url) throws Throwable {

        final CountDownLatch l = new CountDownLatch(1);
        final AtomicReference<String> sessionid = new AtomicReference<String>();

        // on va chercher une sessionID
        client.prepareGet(url).execute(new AsyncCompletionHandlerAdapter() {

            @Override
            public Response onCompleted(Response response) throws Exception {
                try {
                    String body = response.getResponseBody();
                    String array[] = body.split(":");
                    sessionid.set(array[0]);
                } finally {
                    l.countDown();
                }
                return response;
            }
        }).get();

        if (!l.await(TIMEOUT, TimeUnit.SECONDS)) {
            throw new RuntimeException("Timeout out");
        }

        return sessionid.get();
    }

protected void login(final String name, final AsyncHttpClient client, final String url, final String username, final boolean usernameUnique) throws Throwable{
        final CountDownLatch latchGet = new CountDownLatch(2);
        //final CountDownLatch latchGet2 = new CountDownLatch(1);
        final CountDownLatch latchPost = new CountDownLatch(1);

        // fait un connect
        connect(name, client, url);

        //suspend une connection
        suspend(name, client, url, new ResponseListener() {
            @Override
            public void notify(String message) {
                log.info("GET login message received = " + message);

                Assert.assertNotNull(message);

                if(!message.equals("2::")){
                    latchGet.countDown();

                    if(message.charAt(0)==(byte)SocketIOPacketImpl.SOCKETIO_MSG_DELIMITER){
                        System.err.println("Multi-message");
                    }

                    // LOGIN
                    if(latchGet.getCount()==1){
                        if(usernameUnique){
                            //Assert.assertEquals(message, "6:::1+[false]");
                            Assert.assertTrue(message.contains("6:::1+[false]"));
                        } else {
                            //Assert.assertEquals(message, "6:::1+[true]");
                            Assert.assertTrue(message.contains("6:::1+[true]"));
                        }

                    } else {
                        // la liste des users connectes
                        Assert.assertTrue(message.contains("5:::{\"name\":\"nicknames\",\"args\":[{"));
                        //Assert.assertTrue(message.contains("\"" + username + "\":\"" + username + "\""));
                        Assert.assertTrue(message.contains("\"" + username + "\""));
                    }
                }

            } 
        });

        // maintenant on fait login
        sendMessage(name, client, url, "5:1+::{\"name\":\"nickname\",\"args\":[\"" + username + "\"]}", new ResponseListener() {

            boolean found = false;

            @Override
            public void notify(String message) {
                log.info("POST login message received = " + message);
                latchPost.countDown();
                Assert.assertNotNull(message);
                Assert.assertEquals(message, "1");
            }
        });

        if (!latchGet.await(30, TimeUnit.SECONDS)) {
            throw new RuntimeException("Timeout out 1");
        }
        /*
        if (!latchGet2.await(30, TimeUnit.SECONDS)) {

            throw new RuntimeException("Timeout out 2");
        }
        */

        if (!latchPost.await(30, TimeUnit.SECONDS)) {
            throw new RuntimeException("Timeout out 3");
        }
    }

protected void connect(String name, final AsyncHttpClient client, final String url) throws Throwable {
        final CountDownLatch l = new CountDownLatch(1);

        // fait un connect
        suspend(name, client, url, new ResponseListener() {

            @Override
            public void notify(String message) {
                log.info("Connect message received = " + message);
                l.countDown();
                Assert.assertNotNull(message);
                Assert.assertEquals(message, "1::");
            }
        });

        if (!l.await(30, TimeUnit.SECONDS)) {
            throw new RuntimeException("Timeout out");
        }

    }

public static void suspend(final String name, final AsyncHttpClient client, final String url, final ResponseListener listener) throws Throwable {

        System.err.println("Name = " + name + " go in suspend mode");

        Thread t = new Thread() {
            public void run() {
                try {
                    // 
                    client.prepareGet(url).execute(new AsyncCompletionHandlerAdapter() {

                        @Override
                        public com.ning.http.client.AsyncHandler.STATE onBodyPartReceived( HttpResponseBodyPart content) throws Exception {
                            try {
                                String body = new String(content.getBodyPartBytes());
                                System.err.println("suspend Name = " + name + " onBodyPartReceived=" + body);

                                if(listener!=null){
                                    listener.notify(body);
                                } else {
                                    System.err.println("Listner null name=" + name);
                                }

                            } finally {
                            }
                            return super.onBodyPartReceived(content);
                        }

                        @Override
                        public Response onCompleted(Response response) throws Exception {
                            try {
                                String body = response.getResponseBody();
                                System.err.println("suspend Name = " + name + " Body=" + body);
                                if(listener!=null){
                                    listener.notify(body);
                                }
                            } finally {
                            }
                            return response;
                        }
                    }).get();

                } catch (InterruptedException e1) {
                } catch (Exception e) {
                    e.printStackTrace();
                    Assert.fail();
                }
            }
        };

        t.start();

    }

[slandelle]

Will reopen if needed.