support protocol ver 3 and sending byte data

src/main/java/com/github/nkzawa/engineio/client/Socket.java

@@ -13,7 +13,10 @@ import com.google.gson.Gson;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.*;
-import java.util.concurrent.*;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Logger;
 
 
@@ -298,7 +301,7 @@ public abstract class Socket extends Emitter {
                 if (failed[0]) return;
 
                 logger.fine(String.format("probe transport '%s' opened", name));
-                Packet packet = new Packet(Packet.PING, "probe");
+                Packet<String> packet = new Packet<String>(Packet.PING, "probe");
                 transport[0].send(new Packet[] {packet});
                 transport[0].once(Transport.EVENT_PACKET, new Listener() {
                     @Override
@@ -395,7 +398,7 @@ public abstract class Socket extends Emitter {
             this.emit(EVENT_HEARTBEAT);
 
             if (Packet.OPEN.equals(packet.type)) {
-                this.onHandshake(gson.fromJson(packet.data, HandshakeData.class));
+                this.onHandshake(gson.fromJson((String)packet.data, HandshakeData.class));
             } else if (Packet.PONG.equals(packet.type)) {
                 this.setPing();
             } else if (Packet.ERROR.equals(packet.type)) {
@@ -406,7 +409,11 @@ public abstract class Socket extends Emitter {
             } else if (Packet.MESSAGE.equals(packet.type)) {
                 this.emit(EVENT_DATA, packet.data);
                 this.emit(EVENT_MESSAGE, packet.data);
-                this.onmessage(packet.data);
+                if (packet.data instanceof String) {
+                    this.onmessage((String)packet.data);
+                } else if (packet.data instanceof byte[]) {
+                    this.onmessage((byte[])packet.data);
+                }
             }
         } else {
             logger.fine(String.format("packet received with socket readyState '%s'", this.readyState));
@@ -530,6 +537,14 @@ public abstract class Socket extends Emitter {
         this.send(msg, fn);
     }
 
+    public void write(byte[] msg) {
+        this.write(msg, null);
+    }
+
+    public void write(byte[] msg, Runnable fn) {
+        this.send(msg, fn);
+    }
+
     /**
      * Sends a message.
      *
@@ -539,6 +554,10 @@ public abstract class Socket extends Emitter {
         this.send(msg, null);
     }
 
+    public void send(byte[] msg) {
+        this.send(msg, null);
+    }
+
     /**
      * Sends a message.
      *
@@ -554,17 +573,35 @@ public abstract class Socket extends Emitter {
         });
     }
 
+    public void send(final byte[] msg, final Runnable fn) {
+        EventThread.exec(new Runnable() {
+            @Override
+            public void run() {
+                Socket.this.sendPacket(Packet.MESSAGE, msg, fn);
+            }
+        });
+    }
+
     private void sendPacket(String type) {
-        this.sendPacket(type, null, null);
+        this.sendPacket(new Packet(type), null);
     }
 
     private void sendPacket(String type, String data, Runnable fn) {
+        Packet<String> packet = new Packet<String>(type, data);
+        sendPacket(packet, fn);
+    }
+
+    private void sendPacket(String type, byte[] data, Runnable fn) {
+        Packet<byte[]> packet = new Packet<byte[]>(type, data);
+        sendPacket(packet, fn);
+    }
+
+    private void sendPacket(Packet packet, Runnable fn) {
         if (fn == null) {
             // ConcurrentLinkedList does not permit `null`.
             fn = noop;
         }
 
-        Packet packet = new Packet(type, data);
         this.emit(EVENT_PACKET_CREATE, packet);
         this.writeBuffer.offer(packet);
         this.callbackBuffer.offer(fn);
@@ -653,6 +690,8 @@ public abstract class Socket extends Emitter {
         return filteredUpgrades;
     }
 
+    public void onmessage(byte[] data) {};
+
     public abstract void onopen();
 
     public abstract void onmessage(String data);

src/main/java/com/github/nkzawa/engineio/client/Transport.java

@@ -108,6 +108,10 @@ public abstract class Transport extends Emitter {
         this.onPacket(Parser.decodePacket(data));
     }
 
+    protected void onData(byte[] data) {
+        this.onPacket(Parser.decodePacket(data));
+    }
+
     protected void onPacket(Packet packet) {
         this.emit(EVENT_PACKET, packet);
     }

src/main/java/com/github/nkzawa/engineio/client/transports/Polling.java

@@ -1,11 +1,11 @@
 package com.github.nkzawa.engineio.client.transports;
 
 
-import com.github.nkzawa.thread.EventThread;
 import com.github.nkzawa.engineio.client.Transport;
 import com.github.nkzawa.engineio.client.Util;
 import com.github.nkzawa.engineio.parser.Packet;
 import com.github.nkzawa.engineio.parser.Parser;
+import com.github.nkzawa.thread.EventThread;
 
 import java.util.Date;
 import java.util.HashMap;
@@ -94,11 +94,20 @@ abstract public class Polling extends Transport {
         this.emit(EVENT_POLL);
     }
 
+    @Override
     protected void onData(String data) {
+        _onData(data);
+    }
+
+    @Override
+    protected void onData(byte[] data) {
+        _onData(data);
+    }
+
+    private void _onData(Object data) {
         final Polling self = this;
         logger.fine(String.format("polling got data %s", data));
-
+        Parser.DecodePayloadCallback callback = new Parser.DecodePayloadCallback() {
-        Parser.decodePayload(data, new Parser.DecodePayloadCallback() {
             @Override
             public boolean call(Packet packet, int index, int total) {
                 if (self.readyState == ReadyState.OPENING) {
@@ -113,7 +122,13 @@ abstract public class Polling extends Transport {
                 self.onPacket(packet);
                 return true;
             }
-        });
+        };
+
+        if (data instanceof String) {
+            Parser.decodePayload((String)data, callback);
+        } else if (data instanceof byte[]) {
+            Parser.decodePayload((byte[])data, callback);
+        }
 
         if (this.readyState != ReadyState.CLOSED) {
             this.polling = false;
@@ -134,7 +149,7 @@ abstract public class Polling extends Transport {
             @Override
             public void call(Object... args) {
                 logger.fine("writing close packet");
-                self.write(new Packet[] {new Packet(Packet.CLOSE, null)});
+                self.write(new Packet[] {new Packet(Packet.CLOSE)});
             }
         };
 
@@ -152,12 +167,19 @@ abstract public class Polling extends Transport {
     protected void write(Packet[] packets) {
         final Polling self = this;
         this.writable = false;
-        this.doWrite(Parser.encodePayload(packets), new Runnable() {
+        final Runnable callbackfn = new Runnable() {
             @Override
             public void run() {
                 self.writable = true;
                 self.emit(EVENT_DRAIN);
             }
+        };
+
+        Parser.encodePayload(packets, new Parser.EncodeCallback<byte[]>() {
+            @Override
+            public void call(byte[] data) {
+                self.doWrite(data, callbackfn);
+            }
         });
     }
 
@@ -187,7 +209,7 @@ abstract public class Polling extends Transport {
         return schema + "://" + this.hostname + port + this.path + _query;
     }
 
-    abstract protected void doWrite(String data, Runnable fn);
+    abstract protected void doWrite(byte[] data, Runnable fn);
 
     abstract protected void doPoll();
 }

src/main/java/com/github/nkzawa/engineio/client/transports/PollingXHR.java

@@ -7,6 +7,9 @@ import com.github.nkzawa.thread.EventThread;
 import java.io.*;
 import java.net.HttpURLConnection;
 import java.net.URL;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.concurrent.ExecutorService;
@@ -57,7 +60,8 @@ public class PollingXHR extends Polling {
         return req;
     }
 
-    protected void doWrite(String data, final Runnable fn) {
+    @Override
+    protected void doWrite(byte[] data, final Runnable fn) {
         Request.Options opts = new Request.Options();
         opts.method = "POST";
         opts.data = data;
@@ -90,6 +94,7 @@ public class PollingXHR extends Polling {
         this.sendXhr = req;
     }
 
+    @Override
     protected void doPoll() {
         logger.fine("xhr poll");
         Request req = this.request();
@@ -100,8 +105,12 @@ public class PollingXHR extends Polling {
                 EventThread.exec(new Runnable() {
                     @Override
                     public void run() {
-                        String data = args.length > 0 ? (String) args[0] : null;
+                        Object arg = args.length > 0 ? args[0] : null;
-                        self.onData(data);
+                        if (arg instanceof String) {
+                            self.onData((String)arg);
+                        } else if (arg instanceof byte[]) {
+                            self.onData((byte[])arg);
+                        }
                     }
                 });
             }
@@ -134,7 +143,7 @@ public class PollingXHR extends Polling {
 
         String method;
         String uri;
-        String data;
+        byte[] data;
         HttpURLConnection xhr;
 
         public Request(Options opts) {
@@ -159,7 +168,7 @@ public class PollingXHR extends Polling {
 
             if ("POST".equals(this.method)) {
                 xhr.setDoOutput(true);
-                headers.put("Content-type", "text/plain;charset=UTF-8");
+                headers.put("Content-type", "application/octet-stream");
             }
 
             self.onRequestHeaders(headers);
@@ -171,15 +180,15 @@ public class PollingXHR extends Polling {
             xhrService.submit(new Runnable() {
                 @Override
                 public void run() {
-                    BufferedWriter writer = null;
+                    OutputStream output = null;
+                    InputStream input = null;
                     BufferedReader reader = null;
                     try {
                         if (self.data != null) {
-                            byte[] data = self.data.getBytes("UTF-8");
+                            xhr.setFixedLengthStreamingMode(self.data.length);
-                            xhr.setFixedLengthStreamingMode(data.length);
+                            output = new BufferedOutputStream(xhr.getOutputStream());
-                            writer = new BufferedWriter(new OutputStreamWriter(xhr.getOutputStream()));
+                            output.write(self.data);
-                            writer.write(self.data);
+                            output.flush();
-                            writer.flush();
                         }
 
                         Map<String, String> headers = new TreeMap<String, String>(String.CASE_INSENSITIVE_ORDER);
@@ -189,28 +198,46 @@ public class PollingXHR extends Polling {
                         }
                         self.onResponseHeaders(headers);
 
-                        StringBuilder data = null;
-
                         final int statusCode = xhr.getResponseCode();
                         if (HttpURLConnection.HTTP_OK == statusCode) {
-                            String line;
+                            String contentType = xhr.getContentType();
-                            data = new StringBuilder();
+                            if ("application/octet-stream".equalsIgnoreCase(contentType)) {
-                            reader = new BufferedReader(new InputStreamReader(xhr.getInputStream()));
+                                input = new BufferedInputStream(xhr.getInputStream());
-                            while ((line = reader.readLine()) != null) {
+                                List<byte[]> buffers = new ArrayList<byte[]>();
-                                data.append(line);
+                                int capacity = 0;
+                                int len = 0;
+                                byte[] buffer = new byte[1024];
+                                while ((len = input.read(buffer)) > 0) {
+                                    byte[] _buffer = new byte[len];
+                                    System.arraycopy(buffer, 0, _buffer, 0, len);
+                                    buffers.add(_buffer);
+                                    capacity += len;
+                                }
+                                ByteBuffer data = ByteBuffer.allocate(capacity);
+                                for (byte[] b : buffers) {
+                                    data.put(b);
+                                }
+                                self.onData(data.array());
+                            } else {
+                                String line;
+                                StringBuilder data = new StringBuilder();
+                                reader = new BufferedReader(new InputStreamReader(xhr.getInputStream()));
+                                while ((line = reader.readLine()) != null) {
+                                    data.append(line);
+                                }
+                                self.onData(data.toString());
                             }
                         } else {
                             self.onError(new IOException(Integer.toString(statusCode)));
                         }
-
-                        if (data != null) {
-                            self.onData(data.toString());
-                        }
                     } catch (IOException e) {
                         self.onError(e);
                     } finally {
                         try {
-                            if (writer != null) writer.close();
+                            if (output != null) output.close();
+                        } catch (IOException e) {}
+                        try {
+                            if (input != null) input.close();
                         } catch (IOException e) {}
                         try {
                             if (reader != null) reader.close();
@@ -230,6 +257,11 @@ public class PollingXHR extends Polling {
             this.onSuccess();
         }
 
+        private void onData(byte[] data) {
+            this.emit(EVENT_DATA, data);
+            this.onSuccess();
+        }
+
         private void onError(Exception err) {
             this.emit(EVENT_ERROR, err);
             this.cleanup();
@@ -260,7 +292,7 @@ public class PollingXHR extends Polling {
 
             public String uri;
             public String method;
-            public String data;
+            public byte[] data;
         }
     }
 }

src/main/java/com/github/nkzawa/engineio/client/transports/WebSocket.java

@@ -12,6 +12,7 @@ import org.java_websocket.handshake.ServerHandshake;
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -78,6 +79,15 @@ public class WebSocket extends Transport {
                     });
                 }
                 @Override
+                public void onMessage(final ByteBuffer s) {
+                    EventThread.exec(new Runnable() {
+                        @Override
+                        public void run() {
+                            self.onData(s.array());
+                        }
+                    });
+                }
+                @Override
                 public void onError(final Exception e) {
                     EventThread.exec(new Runnable() {
                         @Override
@@ -97,7 +107,16 @@ public class WebSocket extends Transport {
         final WebSocket self = this;
         this.writable = false;
         for (Packet packet : packets) {
-            this.ws.send(Parser.encodePacket(packet));
+            Parser.encodePacket(packet, new Parser.EncodeCallback() {
+                @Override
+                public void call(Object packet) {
+                    if (packet instanceof String) {
+                        self.ws.send((String) packet);
+                    } else if (packet instanceof byte[]) {
+                        self.ws.send((byte[]) packet);
+                    }
+                }
+            });
         }
 
         final Runnable ondrain = new Runnable() {

src/main/java/com/github/nkzawa/engineio/parser/Packet.java

@@ -1,7 +1,7 @@
 package com.github.nkzawa.engineio.parser;
 
 
-public class Packet {
+public class Packet<T> {
 
     static final public String OPEN = "open";
     static final public String CLOSE = "close";
@@ -13,13 +13,14 @@ public class Packet {
     static final public String ERROR = "error";
 
     public String type;
-    public String data;
+    public T data;
+
 
     public Packet(String type) {
         this(type, null);
     }
 
-    public Packet(String type, String data) {
+    public Packet(String type, T data) {
         this.type = type;
         this.data = data;
     }

src/main/java/com/github/nkzawa/engineio/parser/Parser.java

@@ -1,12 +1,17 @@
 package com.github.nkzawa.engineio.parser;
 
 
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 public class Parser {
 
-    public static final int protocol = 2;
+    public static final int protocol = 3;
+
     private static final Map<String, Integer> packets = new HashMap<String, Integer>() {{
         put(Packet.OPEN, 0);
         put(Packet.CLOSE, 1);
@@ -16,54 +21,116 @@ public class Parser {
         put(Packet.UPGRADE, 5);
         put(Packet.NOOP, 6);
     }};
-    private static final Map<Integer, String> bipackets = new HashMap<Integer, String>();
+
+    private static final Map<Integer, String> packetslist = new HashMap<Integer, String>();
     static {
         for (Map.Entry<String, Integer> entry : packets.entrySet()) {
-            bipackets.put(entry.getValue(), entry.getKey());
+            packetslist.put(entry.getValue(), entry.getKey());
         }
     }
 
-    private static Packet err = new Packet(Packet.ERROR, "parser error");
+    private static Packet<String> err = new Packet<String>(Packet.ERROR, "parser error");
 
 
     private Parser() {}
 
-    public static String encodePacket(Packet packet) {
+    public static void encodePacket(Packet packet, EncodeCallback callback) {
+        if (packet.data instanceof byte[]) {
+            @SuppressWarnings("unchecked")
+            Packet<byte[]> _packet = packet;
+            @SuppressWarnings("unchecked")
+            EncodeCallback<byte[]> _callback = callback;
+            encodeByteArray(_packet, _callback);
+            return;
+        }
+
         String encoded = String.valueOf(packets.get(packet.type));
 
-        if (packet.data != null) {
+        if (null != packet.data) {
             encoded += packet.data;
         }
-        return encoded;
+
+        @SuppressWarnings("unchecked")
+        EncodeCallback<String> _callback = callback;
+        _callback.call(encoded);
     }
 
-    public static Packet decodePacket(String data) {
+    private static void encodeByteArray(Packet<byte[]> packet, EncodeCallback<byte[]> callback) {
-        int type = -1;
+        byte[] data = packet.data;
+        byte[] resultArray = new byte[1 + data.length];
+        resultArray[0] = packets.get(packet.type).byteValue();
+        System.arraycopy(data, 0, resultArray, 1, data.length);
+        callback.call(resultArray);
+    }
+
+    public static Packet<String> decodePacket(String data) {
+        int type;
         try {
             type = Character.getNumericValue(data.charAt(0));
-        } catch(IndexOutOfBoundsException e) {}
+        } catch (IndexOutOfBoundsException e) {
-        if (type < 0 || type >= packets.size()) {
+            type = -1;
+        }
+
+        if (type < 0 || type >= packetslist.size()) {
             return err;
         }
 
-        return new Packet(bipackets.get(type), data.length() > 1 ? data.substring(1) : null);
+        if (data.length() > 1) {
+            return new Packet<String>(packetslist.get(type), data.substring(1));
+        } else {
+            return new Packet<String>(packetslist.get(type));
+        }
     }
 
-    public static String encodePayload(Packet[] packets) {
+    public static Packet<byte[]> decodePacket(byte[] data) {
+        int type = data[0];
+        byte[] intArray = new byte[data.length - 1];
+        System.arraycopy(data, 1, intArray, 0, intArray.length);
+        return new Packet<byte[]>(packetslist.get(type), intArray);
+    }
+
+    public static void encodePayload(Packet[] packets, EncodeCallback<byte[]> callback) {
         if (packets.length == 0) {
-            return "0:";
+            callback.call(new byte[0]);
+            return;
         }
 
-        StringBuilder encoded = new StringBuilder();
+        final ArrayList<ByteBuffer> results = new ArrayList<ByteBuffer>(packets.length);
+
         for (Packet packet : packets) {
-            String message = encodePacket(packet);
+            encodePacket(packet, new EncodeCallback() {
-            encoded.append(message.length()).append(":").append(message);
+                @Override
+                public void call(Object packet) {
+                    if (packet instanceof String) {
+                        String encodingLength = String.valueOf(((String)packet).getBytes(Charset.forName("UTF-8")).length);
+                        ByteBuffer sizeBuffer = ByteBuffer.allocate(encodingLength.length() + 2);
+
+                        sizeBuffer.put((byte)0); // is a string
+                        for (char ch : encodingLength.toCharArray()) {
+                            sizeBuffer.put((byte)Character.getNumericValue(ch));
+                        }
+                        sizeBuffer.put((byte)255);
+                        results.add(Buffer.concat(new ByteBuffer[] {sizeBuffer,
+                                ByteBuffer.wrap(((String)packet).getBytes(Charset.forName("UTF-8")))}));
+                        return;
+                    }
+
+                    String encodingLength = String.valueOf(((ByteBuffer)packet).capacity());
+                    ByteBuffer sizeBuffer = ByteBuffer.allocate(encodingLength.length() + 2);
+                    sizeBuffer.put((byte)1); // is binary
+                    for (char ch : encodingLength.toCharArray()) {
+                        sizeBuffer.put((byte)ch);
+                    }
+                    sizeBuffer.put((byte)255);
+                    results.add(Buffer.concat(new ByteBuffer[] {sizeBuffer, (ByteBuffer)packet}));
+                }
+            });
         }
 
-        return encoded.toString();
+        callback.call(Buffer.concat(results.toArray(new ByteBuffer[results.size()])).array());
     }
 
-    public static void decodePayload(String data, DecodePayloadCallback callback) {
+    public static void decodePayload(String data, DecodePayloadCallback<String> callback) {
         if (data == null || data.isEmpty()) {
             callback.call(err, 0, 1);
             return;
@@ -93,7 +160,8 @@ public class Parser {
                 }
 
                 if (msg.length() != 0) {
-                    Packet packet = decodePacket(msg);
+                    Packet<String> packet = decodePacket(msg);
+
                     if (err.type.equals(packet.type) && err.data.equals(packet.data)) {
                         callback.call(err, 0, 1);
                         return;
@@ -113,8 +181,91 @@ public class Parser {
         }
     }
 
-    public static interface DecodePayloadCallback {
+    public static void decodePayload(byte[] data, DecodePayloadCallback callback) {
+        ByteBuffer bufferTail = ByteBuffer.wrap(data);
+        List<Object> buffers = new ArrayList<Object>();
 
-        public boolean call(Packet packet, int index, int total);
+        while (bufferTail.capacity() > 0) {
+            StringBuilder strLen = new StringBuilder();
+            boolean isString = (bufferTail.get(0) & 0xFF) == 0;
+            for (int i = 1; ; i++) {
+                int b = bufferTail.get(i) & 0xFF;
+                if (b == 255) break;
+                strLen.append(b);
+            }
+            bufferTail.position(strLen.length() + 1);
+            bufferTail = bufferTail.slice();
+
+            int msgLength = Integer.parseInt(strLen.toString());
+
+            bufferTail.position(1);
+            bufferTail.limit(msgLength + 1);
+            byte[] msg = new byte[bufferTail.remaining()];
+            bufferTail.get(msg);
+            if (isString) {
+                buffers.add(new String(msg, Charset.forName("UTF-8")));
+            } else {
+                buffers.add(msg);
+            }
+            bufferTail.clear();
+            bufferTail.position(msgLength + 1);
+            bufferTail = bufferTail.slice();
+        }
+
+        int total = buffers.size();
+        for (int i = 0; i < total; i++) {
+            Object buffer = buffers.get(i);
+            if (buffer instanceof String) {
+                @SuppressWarnings("unchecked")
+                DecodePayloadCallback<String> _callback = callback;
+                _callback.call(decodePacket((String)buffer), i, total);
+            } else if (buffer instanceof byte[]) {
+                @SuppressWarnings("unchecked")
+                DecodePayloadCallback<byte[]> _callback = callback;
+                _callback.call(decodePacket((byte[])buffer), i, total);
+            }
+        }
+    }
+
+
+    public static interface EncodeCallback<T> {
+
+        public void call(T data);
+    }
+
+
+    public static interface DecodePayloadCallback<T> {
+
+        public boolean call(Packet<T> packet, int index, int total);
+    }
+}
+
+
+class Buffer {
+
+    private Buffer() {}
+
+    public static ByteBuffer concat(ByteBuffer[] list) {
+        int length = 0;
+        for (ByteBuffer buf : list) {
+            length += buf.capacity();
+        }
+        return concat(list, length);
+    }
+
+    public static ByteBuffer concat(ByteBuffer[] list, int length) {
+        if (list.length == 0) {
+            return ByteBuffer.allocate(0);
+        } else if (list.length == 1) {
+            return list[0];
+        }
+
+        ByteBuffer buffer = ByteBuffer.allocate(length);
+        for (ByteBuffer buf : list) {
+            buf.clear();
+            buffer.put(buf);
+        }
+
+        return buffer;
     }
 }

src/test/java/com/github/nkzawa/engineio/parser/ParserTest.java

@@ -15,91 +15,152 @@ public class ParserTest {
 
     @Test
     public void encodeAsString() {
-        assertThat(encodePacket(new Packet(Packet.MESSAGE, "test")), isA(String.class));
+        encodePacket(new Packet<String>(Packet.MESSAGE, "test"), new EncodeCallback<String>() {
+            @Override
+            public void call(String data) {
+                assertThat(data, isA(String.class));
+            }
+        });
     }
 
     @Test
     public void decodeAsPacket() {
-        assertThat(decodePacket(encodePacket(new Packet(Packet.MESSAGE, "test"))), isA(Packet.class));
+        encodePacket(new Packet<String>(Packet.MESSAGE, "test"), new EncodeCallback<String>() {
+            @Override
+            public void call(String data) {
+                assertThat(decodePacket(data), isA(Packet.class));
+            }
+        });
     }
 
     @Test
     public void noData() {
-        Packet p = decodePacket(encodePacket(new Packet(Packet.MESSAGE)));
+        encodePacket(new Packet(Packet.MESSAGE), new EncodeCallback<String>() {
-        assertThat(p.type, is(Packet.MESSAGE));
+            @Override
-        assertThat(p.data, is(nullValue()));
+            public void call(String data) {
+                Packet p = decodePacket(data);
+                assertThat(p.type, is(Packet.MESSAGE));
+                assertThat(p.data, is(nullValue()));
+            }
+        });
     }
 
     @Test
     public void encodeOpenPacket() {
-        Packet p = decodePacket(encodePacket(new Packet(Packet.OPEN, "{\"some\":\"json\"}")));
+        encodePacket(new Packet<String>(Packet.OPEN, "{\"some\":\"json\"}"), new EncodeCallback<String>() {
-        assertThat(p.type, is(Packet.OPEN));
+            @Override
-        assertThat(p.data, is("{\"some\":\"json\"}"));
+            public void call(String data) {
+                Packet<String> p = decodePacket(data);
+                assertThat(p.type, is(Packet.OPEN));
+                assertThat(p.data, is("{\"some\":\"json\"}"));
+            }
+        });
     }
 
     @Test
     public void encodeClosePacket() {
-        Packet p = decodePacket(encodePacket(new Packet(Packet.CLOSE)));
+        encodePacket(new Packet<String>(Packet.CLOSE), new EncodeCallback<String>() {
-        assertThat(p.type, is(Packet.CLOSE));
+            @Override
+            public void call(String data) {
+                Packet p = decodePacket(data);
+                assertThat(p.type, is(Packet.CLOSE));
+            }
+        });
     }
 
     @Test
     public void encodePingPacket() {
-        Packet p = decodePacket(encodePacket(new Packet(Packet.PING, "1")));
+        encodePacket(new Packet<String>(Packet.PING, "1"), new EncodeCallback<String>() {
-        assertThat(p.type, is(Packet.PING));
+            @Override
-        assertThat(p.data, is("1"));
+            public void call(String data) {
+                Packet<String> p = decodePacket(data);
+                assertThat(p.type, is(Packet.PING));
+                assertThat(p.data, is("1"));
+            }
+        });
     }
 
     @Test
     public void encodePongPacket() {
-        Packet p = decodePacket(encodePacket(new Packet(Packet.PONG, "1")));
+        encodePacket(new Packet<String>(Packet.PONG, "1"), new EncodeCallback<String>() {
-        assertThat(p.type, is(Packet.PONG));
+            @Override
-        assertThat(p.data, is("1"));
+            public void call(String data) {
+                Packet<String> p = decodePacket(data);
+                assertThat(p.type, is(Packet.PONG));
+                assertThat(p.data, is("1"));
+            }
+        });
     }
 
     @Test
     public void encodeMessagePacket() {
-        Packet p = decodePacket(encodePacket(new Packet(Packet.MESSAGE, "aaa")));
+        encodePacket(new Packet<String>(Packet.MESSAGE, "aaa"), new EncodeCallback<String>() {
-        assertThat(p.type, is(Packet.MESSAGE));
+            @Override
-        assertThat(p.data, is("aaa"));
+            public void call(String data) {
+                Packet<String> p = decodePacket(data);
+                assertThat(p.type, is(Packet.MESSAGE));
+                assertThat(p.data, is("aaa"));
+            }
+        });
     }
 
     @Test
     public void encodeUpgradePacket() {
-        Packet p = decodePacket(encodePacket(new Packet(Packet.UPGRADE)));
+        encodePacket(new Packet<String>(Packet.UPGRADE), new EncodeCallback<String>() {
-        assertThat(p.type, is(Packet.UPGRADE));
+            @Override
+            public void call(String data) {
+                Packet p = decodePacket(data);
+                assertThat(p.type, is(Packet.UPGRADE));
+            }
+        });
     }
 
     @Test
     public void encodingFormat() {
-        assertThat(encodePacket(new Packet(Packet.MESSAGE, "test")).matches("[0-9].*"), is(true));
+        encodePacket(new Packet<String>(Packet.MESSAGE, "test"), new EncodeCallback<String>() {
-        assertThat(encodePacket(new Packet(Packet.MESSAGE)).matches("[0-9]"), is(true));
+            @Override
+            public void call(String data) {
+                assertThat(data.matches("[0-9].*"), is(true));
+            }
+        });
+        encodePacket(new Packet<String>(Packet.MESSAGE), new EncodeCallback<String>() {
+            @Override
+            public void call(String data) {
+                assertThat(data.matches("[0-9]"), is(true));
+            }
+        });
     }
 
     @Test
     public void decodeBadFormat() {
-        Packet p = decodePacket(":::");
+        Packet<String> p = decodePacket(":::");
         assertThat(p.type, is(Packet.ERROR));
         assertThat(p.data, is(ERROR_DATA));
     }
 
     @Test
     public void decodeInexistentTypes() {
-        Packet p = decodePacket("94103");
+        Packet<String> p = decodePacket("94103");
         assertThat(p.type, is(Packet.ERROR));
         assertThat(p.data, is(ERROR_DATA));
     }
 
     @Test
-    public void encodePayloadsAsString() {
+    public void encodePayloads() {
-        assertThat(encodePayload(new Packet[] {
+        encodePayload(new Packet[]{new Packet(Packet.PING), new Packet(Packet.PONG)}, new EncodeCallback<byte[]>() {
-                new Packet(Packet.PING), new Packet(Packet.PONG)}), isA(String.class));
+            @Override
+            public void call(byte[] data) {
+                assertThat(data, isA(byte[].class));
+            }
+        });
     }
 
     @Test
     public void encodeAndDecodePayloads() {
-        decodePayload(encodePayload(new Packet[] {new Packet(Packet.MESSAGE, "a")}),
+        encodePayload(new Packet[] {new Packet<String>(Packet.MESSAGE, "a")}, new EncodeCallback<byte[]>() {
-                new DecodePayloadCallback() {
+            @Override
+            public void call(byte[] data) {
+                decodePayload(data, new DecodePayloadCallback() {
                     @Override
                     public boolean call(Packet packet, int index, int total) {
                         boolean isLast = index + 1 == total;
@@ -107,9 +168,12 @@ public class ParserTest {
                         return true;
                     }
                 });
-        decodePayload(encodePayload(new Packet[] {
+            }
-                new Packet(Packet.MESSAGE, "a"), new Packet(Packet.PING)}),
+        });
-                new DecodePayloadCallback() {
+        encodePayload(new Packet[]{new Packet<String>(Packet.MESSAGE, "a"), new Packet(Packet.PING)}, new EncodeCallback<byte[]>() {
+            @Override
+            public void call(byte[] data) {
+                decodePayload(data, new DecodePayloadCallback() {
                     @Override
                     public boolean call(Packet packet, int index, int total) {
                         boolean isLast = index + 1 == total;
@@ -121,26 +185,33 @@ public class ParserTest {
                         return true;
                     }
                 });
+            }
+        });
     }
 
     @Test
     public void encodeAndDecodeEmptyPayloads() {
-        decodePayload(encodePayload(new Packet[] {}), new DecodePayloadCallback() {
+        encodePayload(new Packet[] {}, new EncodeCallback<byte[]>() {
             @Override
-            public boolean call(Packet packet, int index, int total) {
+            public void call(byte[] data) {
-                assertThat(packet.type, is(Packet.OPEN));
+                decodePayload(data, new DecodePayloadCallback() {
-                boolean isLast = index + 1 == total;
+                    @Override
-                assertThat(isLast, is(true));
+                    public boolean call(Packet packet, int index, int total) {
-                return true;
+                        assertThat(packet.type, is(Packet.OPEN));
+                        boolean isLast = index + 1 == total;
+                        assertThat(isLast, is(true));
+                        return true;
+                    }
+                });
             }
         });
     }
 
     @Test
     public void decodePayloadBadFormat() {
-        decodePayload("1!", new DecodePayloadCallback() {
+        decodePayload("1!", new DecodePayloadCallback<String>() {
             @Override
-            public boolean call(Packet packet, int index, int total) {
+            public boolean call(Packet<String> packet, int index, int total) {
                 boolean isLast = index + 1 == total;
                 assertThat(packet.type, is(Packet.ERROR));
                 assertThat(packet.data, is(ERROR_DATA));
@@ -148,9 +219,9 @@ public class ParserTest {
                 return true;
             }
         });
-        decodePayload("", new DecodePayloadCallback() {
+        decodePayload("", new DecodePayloadCallback<String>() {
             @Override
-            public boolean call(Packet packet, int index, int total) {
+            public boolean call(Packet<String> packet, int index, int total) {
                 boolean isLast = index + 1 == total;
                 assertThat(packet.type, is(Packet.ERROR));
                 assertThat(packet.data, is(ERROR_DATA));
@@ -158,9 +229,9 @@ public class ParserTest {
                 return true;
             }
         });
-        decodePayload("))", new DecodePayloadCallback() {
+        decodePayload("))", new DecodePayloadCallback<String>() {
             @Override
-            public boolean call(Packet packet, int index, int total) {
+            public boolean call(Packet<String> packet, int index, int total) {
                 boolean isLast = index + 1 == total;
                 assertThat(packet.type, is(Packet.ERROR));
                 assertThat(packet.data, is(ERROR_DATA));
@@ -172,9 +243,9 @@ public class ParserTest {
 
     @Test
     public void decodePayloadBadLength() {
-        decodePayload("1:", new DecodePayloadCallback() {
+        decodePayload("1:", new DecodePayloadCallback<String>() {
             @Override
-            public boolean call(Packet packet, int index, int total) {
+            public boolean call(Packet<String> packet, int index, int total) {
                 boolean isLast = index + 1 == total;
                 assertThat(packet.type, is(Packet.ERROR));
                 assertThat(packet.data, is(ERROR_DATA));
@@ -186,9 +257,9 @@ public class ParserTest {
 
     @Test
     public void decodePayloadBadPacketFormat() {
-        decodePayload("3:99:", new DecodePayloadCallback() {
+        decodePayload("3:99:", new DecodePayloadCallback<String>() {
             @Override
-            public boolean call(Packet packet, int index, int total) {
+            public boolean call(Packet<String> packet, int index, int total) {
                 boolean isLast = index + 1 == total;
                 assertThat(packet.type, is(Packet.ERROR));
                 assertThat(packet.data, is(ERROR_DATA));
@@ -196,9 +267,9 @@ public class ParserTest {
                 return true;
             }
         });
-        decodePayload("1:aa", new DecodePayloadCallback() {
+        decodePayload("1:aa", new DecodePayloadCallback<String>() {
             @Override
-            public boolean call(Packet packet, int index, int total) {
+            public boolean call(Packet<String> packet, int index, int total) {
                 boolean isLast = index + 1 == total;
                 assertThat(packet.type, is(Packet.ERROR));
                 assertThat(packet.data, is(ERROR_DATA));
@@ -206,9 +277,9 @@ public class ParserTest {
                 return true;
             }
         });
-        decodePayload("1:a2:b", new DecodePayloadCallback() {
+        decodePayload("1:a2:b", new DecodePayloadCallback<String>() {
             @Override
-            public boolean call(Packet packet, int index, int total) {
+            public boolean call(Packet<String> packet, int index, int total) {
                 boolean isLast = index + 1 == total;
                 assertThat(packet.type, is(Packet.ERROR));
                 assertThat(packet.data, is(ERROR_DATA));

src/test/resources/package.json

@@ -3,6 +3,6 @@
   "version": "0.0.0",
   "private": true,
   "dependencies": {
-    "engine.io": "0.8.2"
+    "engine.io": "1.0.5"
   }
 }