Have a config parameter for websocket max buffer size, close AsyncHttpClient#658

Author: Stephane Landelle

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

@@ -133,6 +133,8 @@ public Set<Map.Entry<String, Object>> propertiesSet() {
 
     private NettyWebSocketFactory nettyWebSocketFactory = new DefaultNettyWebSocketFactory();
 
+    private int webSocketMaxBufferSize = 128000000;
+
     public boolean isUseDeadLockChecker() {
         return useDeadLockChecker;
     }
@@ -237,15 +239,23 @@ public void setNettyWebSocketFactory(NettyWebSocketFactory nettyWebSocketFactory
         this.nettyWebSocketFactory = nettyWebSocketFactory;
     }
 
+    public int getWebSocketMaxBufferSize() {
+        return webSocketMaxBufferSize;
+    }
+
+    public void setWebSocketMaxBufferSize(int webSocketMaxBufferSize) {
+        this.webSocketMaxBufferSize = webSocketMaxBufferSize;
+    }
+
     public static interface NettyWebSocketFactory {
-        NettyWebSocket newNettyWebSocket(Channel channel);
+        NettyWebSocket newNettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig);
     }
 
     public class DefaultNettyWebSocketFactory implements NettyWebSocketFactory {
 
         @Override
-        public NettyWebSocket newNettyWebSocket(Channel channel) {
-            return new NettyWebSocket(channel);
+        public NettyWebSocket newNettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig) {
+            return new NettyWebSocket(channel, nettyConfig);
         }
     }
 }

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

@@ -62,7 +62,7 @@ public WebSocketProtocol(ChannelManager channelManager,//
     private void invokeOnSucces(Channel channel, WebSocketUpgradeHandler h) {
         if (!h.touchSuccess()) {
             try {
-                h.onSuccess(nettyConfig.getNettyWebSocketFactory().newNettyWebSocket(channel));
+                h.onSuccess(nettyConfig.getNettyWebSocketFactory().newNettyWebSocket(channel, nettyConfig));
             } catch (Exception ex) {
                 logger.warn("onSuccess unexpected exception", ex);
             }

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

@@ -29,6 +29,7 @@
 import org.slf4j.LoggerFactory;
 
 import com.ning.http.client.HttpResponseBodyPart;
+import com.ning.http.client.providers.netty.NettyAsyncHttpProviderConfig;
 import com.ning.http.client.providers.netty.response.NettyResponseBodyPart;
 import com.ning.http.client.websocket.WebSocket;
 import com.ning.http.client.websocket.WebSocketByteFragmentListener;
@@ -51,19 +52,20 @@ public class NettyWebSocket implements WebSocket {
 
     protected final Channel channel;
     protected final Collection<WebSocketListener> listeners;
-    protected int maxBufferSize = 128000000;
+    protected final int maxBufferSize;
     private int bufferSize;
     private List<ChannelBuffer> _fragments;
     private volatile boolean interestedInByteMessages;
     private volatile boolean interestedInTextMessages;
 
-    public NettyWebSocket(Channel channel) {
-        this(channel, new ConcurrentLinkedQueue<WebSocketListener>());
+    public NettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig) {
+        this(channel, nettyConfig, new ConcurrentLinkedQueue<WebSocketListener>());
     }
 
-    public NettyWebSocket(Channel channel, Collection<WebSocketListener> listeners) {
+    public NettyWebSocket(Channel channel, NettyAsyncHttpProviderConfig nettyConfig, Collection<WebSocketListener> listeners) {
         this.channel = channel;
         this.listeners = listeners;
+        maxBufferSize = nettyConfig.getWebSocketMaxBufferSize();
     }
 
     @Override
@@ -114,14 +116,6 @@ public WebSocket sendPong(byte[] payload) {
         return this;
     }
 
-    public int getMaxBufferSize() {
-        return maxBufferSize;
-    }
-
-    public void setMaxBufferSize(int maxBufferSize) {
-        this.maxBufferSize = Math.max(maxBufferSize, 8192);
-    }
-
     @Override
     public boolean isOpen() {
         return channel.isOpen();