Merge pull request AsyncHttpClient#748 from elakito/work-1.8.x

[1.8.x] support ping/pong listener handling

Author: jfarcand

src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

@@ -95,6 +95,8 @@
 import org.jboss.netty.handler.codec.http.HttpVersion;
 import org.jboss.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
 import org.jboss.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
+import org.jboss.netty.handler.codec.http.websocketx.PingWebSocketFrame;
+import org.jboss.netty.handler.codec.http.websocketx.PongWebSocketFrame;
 import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import org.jboss.netty.handler.codec.http.websocketx.WebSocket08FrameDecoder;
 import org.jboss.netty.handler.codec.http.websocketx.WebSocket08FrameEncoder;
@@ -2274,6 +2276,8 @@ private final class WebSocketProtocol implements Protocol {
         private static final byte OPCODE_CONT = 0x0;
         private static final byte OPCODE_TEXT = 0x1;
         private static final byte OPCODE_BINARY = 0x2;
+        private static final byte OPCODE_PING = 0x9;
+        private static final byte OPCODE_PONG = 0xa;
         private static final byte OPCODE_UNKNOWN = -1;
 
         // We don't need to synchronize as replacing the "ws-decoder" will process using the same thread.
@@ -2376,6 +2380,10 @@ public void handle(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
                     pendingOpcode = OPCODE_TEXT;
                 } else if (frame instanceof BinaryWebSocketFrame) {
                     pendingOpcode = OPCODE_BINARY;
+                } else if (frame instanceof PingWebSocketFrame) {
+                    pendingOpcode = OPCODE_PING;
+                } else if (frame instanceof PongWebSocketFrame) {
+                    pendingOpcode = OPCODE_PONG;
                 }
 
                 HttpChunk webSocketChunk = new HttpChunk() {
@@ -2409,6 +2417,10 @@ public void setContent(ChannelBuffer content) {
                             webSocket.onBinaryFragment(rp.getBodyPartBytes(), frame.isFinalFragment());
                         } else if (pendingOpcode == OPCODE_TEXT) {
                             webSocket.onTextFragment(frame.getBinaryData().toString(UTF8), frame.isFinalFragment());
+                        } else if (pendingOpcode == OPCODE_PING) {
+                            webSocket.onPing(rp.getBodyPartBytes());
+                        } else if (pendingOpcode == OPCODE_PONG) {
+                            webSocket.onPong(rp.getBodyPartBytes());
                         }
 
                         if (frame instanceof CloseWebSocketFrame) {

src/main/java/com/ning/http/client/providers/netty/NettyWebSocket.java

@@ -16,6 +16,8 @@
 import com.ning.http.client.websocket.WebSocketByteListener;
 import com.ning.http.client.websocket.WebSocketCloseCodeReasonListener;
 import com.ning.http.client.websocket.WebSocketListener;
+import com.ning.http.client.websocket.WebSocketPingListener;
+import com.ning.http.client.websocket.WebSocketPongListener;
 import com.ning.http.client.websocket.WebSocketTextListener;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelFutureListener;
@@ -213,6 +215,20 @@ protected void onTextFragment(String message, boolean last) {
         }
     }
 
+    public void onPing(byte[] payload) {
+        for (WebSocketListener listener : listeners) {
+            if (listener instanceof WebSocketPingListener)
+                WebSocketPingListener.class.cast(listener).onPing(payload);
+        }
+    }
+
+    public void onPong(byte[] payload) {
+        for (WebSocketListener listener : listeners) {
+            if (listener instanceof WebSocketPongListener)
+                WebSocketPongListener.class.cast(listener).onPong(payload);
+        }
+    }
+
     protected void onError(Throwable t) {
         for (WebSocketListener l : listeners) {
             try {