Notify Pings and Pongs, close AsyncHttpClient#517

Author: Stephane Landelle

src/main/java/com/ning/http/client/providers/netty/handler/WebSocketProtocol.java

@@ -23,6 +23,9 @@
 import org.jboss.netty.handler.codec.http.HttpResponse;
 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.WebSocketFrame;
 
 import com.ning.http.client.AsyncHandler.STATE;
@@ -131,7 +134,7 @@ public void handle(Channel channel, NettyResponseFuture<?> future, Object e) thr
                     Channels.setDiscard(channel);
                     CloseWebSocketFrame closeFrame = CloseWebSocketFrame.class.cast(frame);
                     webSocket.onClose(closeFrame.getStatusCode(), closeFrame.getReasonText());
-
+                    
                 } else if (frame.getBinaryData() != null) {
                     HttpChunk webSocketChunk = new HttpChunk() {
                         private ChannelBuffer content = frame.getBinaryData();
@@ -152,13 +155,17 @@ public void setContent(ChannelBuffer content) {
                         }
                     };
 
-                    NettyResponseBodyPart rp = new NettyResponseBodyPart(null, webSocketChunk, frame.isFinalFragment());
-                    handler.onBodyPartReceived(rp);
+                    NettyResponseBodyPart part = new NettyResponseBodyPart(null, webSocketChunk, frame.isFinalFragment());
+                    handler.onBodyPartReceived(part);
 
                     if (frame instanceof BinaryWebSocketFrame) {
-                        webSocket.onBinaryFragment(rp);
-                    } else {
-                        webSocket.onTextFragment(rp);
+                        webSocket.onBinaryFragment(part);
+                    } else if (frame instanceof TextWebSocketFrame) {
+                        webSocket.onTextFragment(part);
+                    } else if (frame instanceof PingWebSocketFrame) {
+                        webSocket.onPing(part);
+                    } else if (frame instanceof PongWebSocketFrame) {
+                        webSocket.onPong(part);
                     }
                 }
             } else {

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

@@ -35,6 +35,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.WebSocketTextFragmentListener;
 import com.ning.http.client.websocket.WebSocketTextListener;
 
@@ -297,4 +299,20 @@ public void onTextFragment(HttpResponseBodyPart part) {
                 bufferFragment(fragment);
         }
     }
+
+    public void onPing(HttpResponseBodyPart part) {
+        for (WebSocketListener listener : listeners) {
+            if (listener instanceof WebSocketPingListener)
+                // bytes are cached in the part
+                WebSocketPingListener.class.cast(listener).onPing(part.getBodyPartBytes());
+        }
+    }
+
+    public void onPong(HttpResponseBodyPart part) {
+        for (WebSocketListener listener : listeners) {
+            if (listener instanceof WebSocketPongListener)
+                // bytes are cached in the part
+                WebSocketPongListener.class.cast(listener).onPong(part.getBodyPartBytes());
+        }
+    }
 }