fix tests

src/test/java/com/github/nkzawa/socketio/client/ConnectionTest.java

@@ -13,7 +13,8 @@ import java.nio.charset.Charset;
 import java.util.Date;
 import java.util.Timer;
 import java.util.TimerTask;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.is;
@@ -26,7 +27,7 @@ public class ConnectionTest extends Connection {
 
     @Test(timeout = TIMEOUT)
     public void connectToLocalhost() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
@@ -35,19 +36,19 @@ public class ConnectionTest extends Connection {
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        socket.close();
+                        values.offer("done");
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.connect();
-        latch.await();
+        values.take();
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void workWithAcks() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
@@ -72,8 +73,7 @@ public class ConnectionTest extends Connection {
                         JSONObject data = (JSONObject)args[1];
                         try {
                             if ((Integer)args[0] == 5 && data.getBoolean("test")) {
-                                socket.close();
+                                values.offer("done");
-                                latch.countDown();
                             }
                         } catch (JSONException e) {
                             throw new AssertionError(e);
@@ -83,12 +83,13 @@ public class ConnectionTest extends Connection {
             }
         });
         socket.connect();
-        latch.await();
+        values.take();
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void receiveDateWithAck() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
@@ -97,9 +98,7 @@ public class ConnectionTest extends Connection {
                     socket.emit("getAckDate", new JSONObject("{test: true}"), new Ack() {
                         @Override
                         public void call(Object... args) {
-                            assertThat(args[0], instanceOf(String.class));
+                            values.offer(args[0]);
-                            socket.close();
-                            latch.countDown();
                         }
                     });
                 } catch (JSONException e) {
@@ -108,12 +107,13 @@ public class ConnectionTest extends Connection {
             }
         });
         socket.connect();
-        latch.await();
+        assertThat(values.take(), instanceOf(String.class));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void workWithFalse() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
@@ -122,20 +122,19 @@ public class ConnectionTest extends Connection {
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        assertThat((Boolean)args[0], is(false));
+                        values.offer(args[0]);
-                        socket.close();
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.connect();
-        latch.await();
+        assertThat((Boolean)values.take(), is(false));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void receiveUTF8MultibyteCharacters() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         final String[] correct = new String[] {
             "てすと",
             "Я Б Г Д Ж Й",
@@ -152,12 +151,7 @@ public class ConnectionTest extends Connection {
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        assertThat((String)args[0], is(correct[i[0]]));
+                        values.offer(args[0]);
-                        i[0]++;
-                        if (i[0] == correct.length) {
-                            socket.close();
-                            latch.countDown();
-                        }
                     }
                 });
                 for (String data : correct) {
@@ -166,12 +160,15 @@ public class ConnectionTest extends Connection {
             }
         });
         socket.connect();
-        latch.await();
+        for (String expected : correct) {
+            assertThat((String)values.take(), is(expected));
+        }
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void connectToNamespaceAfterConnectionEstablished() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         final Manager manager = new Manager(new URI(uri()));
         socket = manager.socket("/");
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
@@ -181,27 +178,26 @@ public class ConnectionTest extends Connection {
                 foo.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
+                        values.offer("done");
                         foo.close();
-                        socket.close();
-                        latch.countDown();
                     }
                 });
                 foo.open();
             }
         });
         socket.open();
-        latch.await();
+        values.take();
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void reconnectByDefault() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         socket = client();
         socket.io().on(Manager.EVENT_RECONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... objects) {
-                socket.close();
+                values.offer("done");
-                latch.countDown();
             }
         });
         socket.open();
@@ -211,18 +207,18 @@ public class ConnectionTest extends Connection {
                 socket.io().engine.close();
             }
         }, 500);
-        latch.await();
+        values.take();
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void reconnectEventFireInSocket() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         socket = client();
         socket.on(Socket.EVENT_RECONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... objects) {
-                socket.close();
+                values.offer("done");
-                latch.countDown();
             }
         });
         socket.open();
@@ -232,12 +228,13 @@ public class ConnectionTest extends Connection {
                 socket.io().engine.close();
             }
         }, 500);
-        latch.await();
+        values.take();
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void tryToReconnectTwiceAndFailWithIncorrectAddress() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         IO.Options opts = new IO.Options();
         opts.reconnection = true;
         opts.reconnectionAttempts = 2;
@@ -257,19 +254,18 @@ public class ConnectionTest extends Connection {
         manager.on(Manager.EVENT_RECONNECT_FAILED, new Emitter.Listener() {
             @Override
             public void call(Object... objects) {
-                assertThat(reconnects[0], is(2));
+                values.offer(reconnects[0]);
-                socket.close();
-                latch.countDown();
             }
         });
 
         socket.open();
-        latch.await();
+        assertThat((Integer)values.take(), is(2));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void tryToReconnectTwiceAndFailWithImmediateTimeout() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         IO.Options opts = new IO.Options();
         opts.reconnection = true;
         opts.timeout = 0;
@@ -289,20 +285,19 @@ public class ConnectionTest extends Connection {
         manager.on(Manager.EVENT_RECONNECT_FAILED, new Emitter.Listener() {
             @Override
             public void call(Object... objects) {
-                assertThat(reconnects[0], is(2));
+                values.offer(reconnects[0]);
-                socket.close();
-                latch.countDown();
             }
         });
 
         socket = manager.socket("/timeout");
         socket.open();
-        latch.await();
+        assertThat((Integer)values.take(), is(2));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void notTryToReconnectWithIncorrectPortWhenReconnectionDisabled() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         IO.Options opts = new IO.Options();
         opts.reconnection = false;
         Manager manager = new Manager(new URI("http://localhost:9823"), opts);
@@ -321,8 +316,7 @@ public class ConnectionTest extends Connection {
                 timer.schedule(new TimerTask() {
                     @Override
                     public void run() {
-                        socket.close();
+                        values.offer("done");
-                        latch.countDown();
                     }
                 }, 1000);
             }
@@ -330,12 +324,13 @@ public class ConnectionTest extends Connection {
 
         socket = manager.socket("/invalid");
         socket.open();
-        latch.await();
+        values.take();
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void fireReconnectEventsOnSocket() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         Manager.Options opts = new Manager.Options();
         opts.reconnection = true;
@@ -350,7 +345,7 @@ public class ConnectionTest extends Connection {
             @Override
             public void call(Object... args) {
                 reconnects[0]++;
-                assertThat((Integer)args[0], is(reconnects[0]));
+                values.offer(args[0]);
             }
         };
 
@@ -358,18 +353,18 @@ public class ConnectionTest extends Connection {
         socket.on(Socket.EVENT_RECONNECT_FAILED, new Emitter.Listener() {
             @Override
             public void call(Object... objects) {
-                assertThat(reconnects[0], is(2));
+                values.offer(reconnects[0]);
-                socket.close();
-                latch.countDown();
             }
         });
         socket.open();
-        latch.await();
+        assertThat((Integer)values.take(), is(reconnects[0]));
+        assertThat((Integer)values.take(), is(2));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void fireReconnectingWithAttemptsNumberWhenReconnectingTwice() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         Manager.Options opts = new Manager.Options();
         opts.reconnection = true;
@@ -384,7 +379,7 @@ public class ConnectionTest extends Connection {
             @Override
             public void call(Object... args) {
                 reconnects[0]++;
-                assertThat((Integer)args[0], is(reconnects[0]));
+                values.offer(args[0]);
             }
         };
 
@@ -392,18 +387,18 @@ public class ConnectionTest extends Connection {
         socket.on(Socket.EVENT_RECONNECT_FAILED, new Emitter.Listener() {
             @Override
             public void call(Object... objects) {
-                assertThat(reconnects[0], is(2));
+                values.offer(reconnects[0]);
-                socket.close();
-                latch.countDown();
             }
         });
         socket.open();
-        latch.await();
+        assertThat((Integer)values.take(), is(reconnects[0]));
+        assertThat((Integer)values.take(), is(2));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void emitDateAsString() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
@@ -412,20 +407,20 @@ public class ConnectionTest extends Connection {
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        assertThat(args[0], instanceOf(String.class));
+                        values.offer(args[0]);
                         socket.close();
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.connect();
-        latch.await();
+        assertThat(values.take(), instanceOf(String.class));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
-    public void emitDateInObject() throws URISyntaxException, InterruptedException {
+    public void emitDateInObject() throws URISyntaxException, InterruptedException, JSONException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
@@ -440,54 +435,48 @@ public class ConnectionTest extends Connection {
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        assertThat(args[0], instanceOf(JSONObject.class));
+                        values.offer(args[0]);
-                        try {
-                            assertThat(((JSONObject)args[0]).get("date"), instanceOf(String.class));
-                        } catch (JSONException e) {
-                            throw new AssertionError(e);
-                        }
-                        socket.close();
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.connect();
-        latch.await();
+        Object data = values.take();
+        assertThat(data, instanceOf(JSONObject.class));
+        assertThat(((JSONObject)data).get("date"), instanceOf(String.class));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void sendAndGetBinaryData() throws URISyntaxException, InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
+        final byte[] buf = "asdfasdf".getBytes(Charset.forName("UTF-8"));
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... args) {
-                final byte[] buf = "asdfasdf".getBytes(Charset.forName("UTF-8"));
                 socket.emit("echo", buf);
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        assertThat(args[0], instanceOf(byte[].class));
+                        values.offer(args[0]);
-                        assertThat((byte[])args[0], is(buf));
-                        socket.close();
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.open();
-        latch.await();
+        assertThat((byte[])values.take(), is(buf));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
-    public void sendBinaryDataMixedWithJson() throws URISyntaxException, InterruptedException {
+    public void sendBinaryDataMixedWithJson() throws URISyntaxException, InterruptedException, JSONException {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
+        final byte[] buf = "howdy".getBytes(Charset.forName("UTF-8"));
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... args) {
-                final byte[] buf = "howdy".getBytes(Charset.forName("UTF-8"));
                 JSONObject data = new JSONObject();
                 try {
                     data.put("hello", "lol");
@@ -500,55 +489,41 @@ public class ConnectionTest extends Connection {
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        JSONObject a = (JSONObject)args[0];
+                        values.offer(args[0]);
-                        try {
-                            assertThat(a.getString("hello"), is("lol"));
-                            assertThat((byte[])a.get("message"), is(buf));
-                            assertThat(a.getString("goodbye"), is("gotcha"));
-                        } catch (JSONException e) {
-                            throw new AssertionError(e);
-                        }
-                        socket.close();
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.open();
-        latch.await();
+        JSONObject a = (JSONObject)values.take();
+        assertThat(a.getString("hello"), is("lol"));
+        assertThat((byte[])a.get("message"), is(buf));
+        assertThat(a.getString("goodbye"), is("gotcha"));
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void sendEventsWithByteArraysInTheCorrectOrder() throws Exception {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
+        final byte[] buf = "abuff1".getBytes(Charset.forName("UTF-8"));
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... args) {
-                final byte[] buf = "abuff1".getBytes(Charset.forName("UTF-8"));
                 socket.emit("echo", buf);
                 socket.emit("echo", "please arrive second");
 
-                final boolean[] receivedAbuff1 = new boolean[] {false};
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        Object data = args[0];
+                        values.offer(args[0]);
-                        if (data instanceof byte[]) {
-                            assertThat((byte[])data, is(buf));
-                            receivedAbuff1[0] = true;
-                            return;
-                        }
-
-                        assertThat((String)data, is("please arrive second"));
-                        assertThat(receivedAbuff1[0], is(true));
-                        socket.close();
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.open();
-        latch.await();
+        assertThat((byte[])values.take(), is(buf));
+        assertThat((String)values.take(), is("please arrive second"));
+        socket.close();
     }
 }

src/test/java/com/github/nkzawa/socketio/client/SSLConnectionTest.java

@@ -14,7 +14,8 @@ import java.io.FileInputStream;
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.security.KeyStore;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 @RunWith(JUnit4.class)
 public class SSLConnectionTest extends Connection {
@@ -71,7 +72,7 @@ public class SSLConnectionTest extends Connection {
 
     @Test(timeout = TIMEOUT)
     public void connect() throws Exception {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         IO.Options opts = createOptions();
         opts.sslContext = createSSLContext();
         socket = client(opts);
@@ -82,19 +83,19 @@ public class SSLConnectionTest extends Connection {
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        socket.close();
+                        values.offer("done");
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.connect();
-        latch.await();
+        values.take();
+        socket.close();
     }
 
     @Test(timeout = TIMEOUT)
     public void defaultSSLContext() throws Exception {
-        final CountDownLatch latch = new CountDownLatch(1);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
         IO.setDefaultSSLContext(createSSLContext());
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
@@ -104,13 +105,13 @@ public class SSLConnectionTest extends Connection {
                 socket.on("echoBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        socket.close();
+                        values.offer("done");
-                        latch.countDown();
                     }
                 });
             }
         });
         socket.connect();
-        latch.await();
+        values.take();
+        socket.close();
     }
 }

src/test/java/com/github/nkzawa/socketio/client/ServerConnectionTest.java

@@ -13,7 +13,6 @@ import java.net.URISyntaxException;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
 
 import static org.hamcrest.CoreMatchers.*;
 import static org.junit.Assert.assertThat;
@@ -26,31 +25,32 @@ public class ServerConnectionTest extends Connection {
 
     @Test(timeout = TIMEOUT)
     public void openAndClose() throws URISyntaxException, InterruptedException {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... args) {
-                assertThat(args.length, is(0));
+                values.offer(args);
                 socket.disconnect();
             }
         }).on(Socket.EVENT_DISCONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... args) {
-                assertThat(args.length, is(1));
+                values.offer(args);
-                assertThat(args[0], is(instanceOf(String.class)));
-                semaphore.release();
             }
         });
         socket.connect();
-        semaphore.acquire();
+
+        assertThat(((Object[])values.take()).length, is(0));
+        Object[] args = (Object[] )values.take();
+        assertThat(args.length, is(1));
+        assertThat(args[0], is(instanceOf(String.class)));
     }
 
     @Test(timeout = TIMEOUT)
     public void message() throws URISyntaxException, InterruptedException {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
-        final int[] count = new int[] {0};
 
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
@@ -60,26 +60,20 @@ public class ServerConnectionTest extends Connection {
             }
         }).on(Socket.EVENT_MESSAGE, new Emitter.Listener() {
             @Override
-            public void call(Object... objects) {
+            public void call(Object... args) {
-                switch (count[0]++) {
+                values.offer(args);
-                case 0:
-                    assertThat(objects, is(new Object[] {"hello client"}));
-                    break;
-                case 1:
-                    assertThat(objects, is(new Object[] {"foo", "bar"}));
-                    socket.disconnect();
-                    semaphore.release();
-                    break;
-                }
             }
         });
         socket.connect();
-        semaphore.acquire();
+
+        assertThat((Object[])values.take(), is(new Object[] {"hello client"}));
+        assertThat((Object[])values.take(), is(new Object[] {"foo", "bar"}));
+        socket.disconnect();
     }
 
     @Test(timeout = TIMEOUT)
     public void event() throws Exception {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         final JSONObject obj = new JSONObject();
         obj.put("foo", 1);
@@ -88,28 +82,27 @@ public class ServerConnectionTest extends Connection {
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... objects) {
-                System.out.println("connect:");
                 socket.emit("echo", obj, null, "bar");
             }
         }).on("echoBack", new Emitter.Listener() {
             @Override
             public void call(Object... args) {
-                System.out.println(String.format("echoBack: %s, %s, %s", args));
+                values.offer(args);
-                assertThat(args.length, is(3));
-                assertThat(args[0].toString(), is(obj.toString()));
-                assertThat(args[1], is(nullValue()));
-                assertThat((String)args[2], is("bar"));
-                socket.disconnect();
-                semaphore.release();
             }
         });
         socket.connect();
-        semaphore.acquire();
+
+        Object[] args = (Object[])values.take();
+        assertThat(args.length, is(3));
+        assertThat(args[0].toString(), is(obj.toString()));
+        assertThat(args[1], is(nullValue()));
+        assertThat((String)args[2], is("bar"));
+        socket.disconnect();
     }
 
     @Test(timeout = TIMEOUT)
     public void ack() throws Exception {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         final JSONObject obj = new JSONObject();
         obj.put("foo", 1);
@@ -118,27 +111,26 @@ public class ServerConnectionTest extends Connection {
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
             @Override
             public void call(Object... objects) {
-                System.out.println("connect:");
                 socket.emit("ack", new Object[] {obj, "bar"}, new Ack() {
                     @Override
                     public void call(Object... args) {
-                        System.out.println(String.format("ack: %s, %s", args));
+                        values.offer(args);
-                        assertThat(args.length, is(2));
-                        assertThat(args[0].toString(), is(obj.toString()));
-                        assertThat((String)args[1], is("bar"));
-                        socket.disconnect();
-                        semaphore.release();
                     }
                 });
             }
         });
         socket.connect();
-        semaphore.acquire();
+
+        Object[] args = (Object[])values.take();
+        assertThat(args.length, is(2));
+        assertThat(args[0].toString(), is(obj.toString()));
+        assertThat((String)args[1], is("bar"));
+        socket.disconnect();
     }
 
     @Test(timeout = TIMEOUT)
     public void ackWithoutArgs() throws URISyntaxException, InterruptedException {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
@@ -147,20 +139,20 @@ public class ServerConnectionTest extends Connection {
                 socket.emit("ack", null, new Ack() {
                     @Override
                     public void call(Object... args) {
-                        assertThat(args.length, is(0));
+                        values.offer(args.length);
-                        socket.disconnect();
-                        semaphore.release();
                     }
                 });
             }
         });
         socket.connect();
-        semaphore.acquire();
+
+        assertThat((Integer)values.take(), is(0));
+        socket.disconnect();
     }
 
     @Test(timeout = TIMEOUT)
     public void ackWithoutArgsFromClient() throws URISyntaxException, InterruptedException {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
@@ -169,29 +161,33 @@ public class ServerConnectionTest extends Connection {
                 socket.on("ack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        assertThat(args.length, is(1));
+                        values.offer(args);
-                        assertThat(args[0], is(instanceOf(Ack.class)));
                         Ack ack = (Ack)args[0];
                         ack.call();
                     }
                 }).on("ackBack", new Emitter.Listener() {
                     @Override
                     public void call(Object... args) {
-                        assertThat(args.length, is(0));
+                        values.offer(args);
                         socket.disconnect();
-                        semaphore.release();
                     }
                 });
                 socket.emit("callAck");
             }
         });
         socket.connect();
-        semaphore.acquire();
+
+        Object[] args = (Object[])values.take();
+        assertThat(args.length, is(1));
+        assertThat(args[0], is(instanceOf(Ack.class)));
+        args = (Object[])values.take();
+        assertThat(args.length, is(0));
+        socket.disconnect();
     }
 
     @Test(timeout = TIMEOUT)
     public void closeEngineConnection() throws URISyntaxException, InterruptedException {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
@@ -200,19 +196,19 @@ public class ServerConnectionTest extends Connection {
                 socket.io().engine.on(com.github.nkzawa.engineio.client.Socket.EVENT_CLOSE, new Emitter.Listener() {
                     @Override
                     public void call(Object... objects) {
-                        semaphore.release();
+                        values.offer("done");
                     }
                 });
                 socket.disconnect();
             }
         });
         socket.connect();
-        semaphore.acquire();
+        values.take();
     }
 
     @Test(timeout = TIMEOUT)
     public void broadcast() throws URISyntaxException, InterruptedException {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
@@ -234,21 +230,22 @@ public class ServerConnectionTest extends Connection {
             }
         }).on("broadcastBack", new Emitter.Listener() {
             @Override
-            public void call(Object... objects) {
+            public void call(Object... args) {
-                assertThat(objects.length, is(1));
+                values.offer(args);
-                assertThat((String)objects[0], is("hi"));
-                socket.disconnect();
-                socket2.disconnect();
-                semaphore.release();
             }
         });
         socket.connect();
-        semaphore.acquire();
+
+        Object[] args = (Object[])values.take();
+        assertThat(args.length, is(1));
+        assertThat((String)args[0], is("hi"));
+        socket.disconnect();
+        socket2.disconnect();
     }
 
     @Test(timeout = TIMEOUT)
     public void room() throws URISyntaxException, InterruptedException {
-        final Semaphore semaphore = new Semaphore(0);
+        final BlockingQueue<Object> values = new LinkedBlockingQueue<Object>();
 
         socket = client();
         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
@@ -258,15 +255,16 @@ public class ServerConnectionTest extends Connection {
             }
         }).on("roomBack", new Emitter.Listener() {
             @Override
-            public void call(Object... objects) {
+            public void call(Object... args) {
-                assertThat(objects.length, is(1));
+                values.offer(args);
-                assertThat((String)objects[0], is("hi"));
-                socket.disconnect();
-                semaphore.release();
             }
         });
         socket.connect();
-        semaphore.acquire();
+
+        Object[] args = (Object[])values.take();
+        assertThat(args.length, is(1));
+        assertThat((String)args[0], is("hi"));
+        socket.disconnect();
     }
 
     @Test(timeout = TIMEOUT)