Skip to content

Commit d5b88ee

Browse files
rlubkerlubke
committed
Grizzly-related changes for #207. By default, fragments will be buffered. Once the last fragment has been received the full message will be passed to the listener.
Original functionality can be re-enabled by setting the BUFFER_WEBSOCKET_FRAGMENTS to false on the GrizzlyAsyncHttpProviderConfig instance.
1 parent 3cf8b09 commit d5b88ee

File tree

2 files changed

+74
-13
lines changed

2 files changed

+74
-13
lines changed

‎src/main/java/com/ning/http/client/providers/grizzly/GrizzlyAsyncHttpProvider.java

Lines changed: 67 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import com.ning.http.client.AsyncHandler;
1717
import com.ning.http.client.AsyncHttpClientConfig;
1818
import com.ning.http.client.AsyncHttpProvider;
19+
import com.ning.http.client.AsyncHttpProviderConfig;
1920
import com.ning.http.client.Body;
2021
import com.ning.http.client.BodyGenerator;
2122
import com.ning.http.client.ConnectionPoolKeyStrategy;
@@ -111,6 +112,7 @@
111112
import org.slf4j.LoggerFactory;
112113

113114
import javax.net.ssl.SSLContext;
115+
import java.io.ByteArrayOutputStream;
114116
import java.io.EOFException;
115117
import java.io.File;
116118
import java.io.FileInputStream;
@@ -135,6 +137,7 @@
135137
import java.util.concurrent.atomic.AtomicInteger;
136138
import java.util.concurrent.atomic.AtomicLong;
137139

140+
import static com.ning.http.client.providers.grizzly.GrizzlyAsyncHttpProviderConfig.Property.BUFFER_WEBSOCKET_FRAGMENTS;
138141
import static com.ning.http.client.providers.grizzly.GrizzlyAsyncHttpProviderConfig.Property.MAX_HTTP_PACKET_HEADER_SIZE;
139142
import static com.ning.http.client.providers.grizzly.GrizzlyAsyncHttpProviderConfig.Property.TRANSPORT_CUSTOMIZER;
140143

@@ -1316,8 +1319,9 @@ protected void onHttpHeadersParsed(HttpHeader httpHeader,
13161319
if (context.isWSRequest) {
13171320
try {
13181321
context.protocolHandler.setConnection(ctx.getConnection());
1319-
DefaultWebSocket ws = new DefaultWebSocket(context.protocolHandler);
1320-
context.webSocket = new GrizzlyWebSocketAdapter(ws);
1322+
final GrizzlyWebSocketAdapter webSocketAdapter = createWebSocketAdapter(context);
1323+
context.webSocket = webSocketAdapter;
1324+
DefaultWebSocket ws = webSocketAdapter.gWebSocket;
13211325
if (context.currentState == AsyncHandler.STATE.UPGRADE) {
13221326
httpHeader.setChunked(false);
13231327
ws.onConnect();
@@ -1409,6 +1413,16 @@ protected boolean onHttpPacketParsed(HttpHeader httpHeader, FilterChainContext c
14091413

14101414
// ----------------------------------------------------- Private Methods
14111415

1416+
private static GrizzlyWebSocketAdapter createWebSocketAdapter(final HttpTransactionContext context) {
1417+
DefaultWebSocket ws = new DefaultWebSocket(context.protocolHandler);
1418+
AsyncHttpProviderConfig config = context.provider.clientConfig.getAsyncHttpProviderConfig();
1419+
boolean bufferFragments = true;
1420+
if (config instanceof GrizzlyAsyncHttpProviderConfig) {
1421+
bufferFragments = (Boolean) ((GrizzlyAsyncHttpProviderConfig) config).getProperty(BUFFER_WEBSOCKET_FRAGMENTS);
1422+
}
1423+
1424+
return new GrizzlyWebSocketAdapter(ws, bufferFragments);
1425+
}
14121426

14131427
private static boolean isRedirectAllowed(final HttpTransactionContext ctx) {
14141428
boolean allowed = ctx.request.isRedirectEnabled();
@@ -2619,13 +2633,16 @@ public void getBytes(byte[] bytes) {
26192633

26202634
private static final class GrizzlyWebSocketAdapter implements WebSocket {
26212635

2622-
private final org.glassfish.grizzly.websockets.WebSocket gWebSocket;
2636+
final DefaultWebSocket gWebSocket;
2637+
final boolean bufferFragments;
26232638

26242639
// -------------------------------------------------------- Constructors
26252640

26262641

2627-
GrizzlyWebSocketAdapter(final org.glassfish.grizzly.websockets.WebSocket gWebSocket) {
2628-
this.gWebSocket = gWebSocket;
2642+
GrizzlyWebSocketAdapter(final DefaultWebSocket gWebSocket,
2643+
final boolean bufferFragments) {
2644+
this.gWebSocket = gWebSocket;
2645+
this.bufferFragments = bufferFragments;
26292646
}
26302647

26312648

@@ -2706,14 +2723,25 @@ public void close() {
27062723
private static final class AHCWebSocketListenerAdapter implements org.glassfish.grizzly.websockets.WebSocketListener {
27072724

27082725
private final WebSocketListener ahcListener;
2709-
private final WebSocket webSocket;
2726+
private final GrizzlyWebSocketAdapter webSocket;
2727+
private final StringBuilder stringBuffer;
2728+
private final ByteArrayOutputStream byteArrayOutputStream;
2729+
27102730

27112731
// -------------------------------------------------------- Constructors
27122732

27132733

2714-
AHCWebSocketListenerAdapter(final WebSocketListener ahcListener, WebSocket webSocket) {
2734+
AHCWebSocketListenerAdapter(final WebSocketListener ahcListener,
2735+
final GrizzlyWebSocketAdapter webSocket) {
27152736
this.ahcListener = ahcListener;
27162737
this.webSocket = webSocket;
2738+
if (webSocket.bufferFragments) {
2739+
stringBuffer = new StringBuilder();
2740+
byteArrayOutputStream = new ByteArrayOutputStream();
2741+
} else {
2742+
stringBuffer = null;
2743+
byteArrayOutputStream = null;
2744+
}
27172745
}
27182746

27192747

@@ -2788,21 +2816,47 @@ public void onPong(org.glassfish.grizzly.websockets.WebSocket webSocket, byte[]
27882816
}
27892817

27902818
@Override
2791-
public void onFragment(org.glassfish.grizzly.websockets.WebSocket webSocket, String s, boolean b) {
2819+
public void onFragment(org.glassfish.grizzly.websockets.WebSocket webSocket, String s, boolean last) {
27922820
try {
2793-
if (WebSocketTextListener.class.isAssignableFrom(ahcListener.getClass())) {
2794-
WebSocketTextListener.class.cast(ahcListener).onFragment(s, b);
2821+
if (this.webSocket.bufferFragments) {
2822+
synchronized (this.webSocket) {
2823+
stringBuffer.append(s);
2824+
if (last) {
2825+
if (WebSocketTextListener.class.isAssignableFrom(ahcListener.getClass())) {
2826+
final String message = stringBuffer.toString();
2827+
stringBuffer.setLength(0);
2828+
WebSocketTextListener.class.cast(ahcListener).onMessage(message);
2829+
}
2830+
}
2831+
}
2832+
} else {
2833+
if (WebSocketTextListener.class.isAssignableFrom(ahcListener.getClass())) {
2834+
WebSocketTextListener.class.cast(ahcListener).onFragment(s, last);
2835+
}
27952836
}
27962837
} catch (Throwable e) {
27972838
ahcListener.onError(e);
27982839
}
27992840
}
28002841

28012842
@Override
2802-
public void onFragment(org.glassfish.grizzly.websockets.WebSocket webSocket, byte[] bytes, boolean b) {
2843+
public void onFragment(org.glassfish.grizzly.websockets.WebSocket webSocket, byte[] bytes, boolean last) {
28032844
try {
2804-
if (WebSocketByteListener.class.isAssignableFrom(ahcListener.getClass())) {
2805-
WebSocketByteListener.class.cast(ahcListener).onFragment(bytes, b);
2845+
if (this.webSocket.bufferFragments) {
2846+
synchronized (this.webSocket) {
2847+
byteArrayOutputStream.write(bytes);
2848+
if (last) {
2849+
if (WebSocketByteListener.class.isAssignableFrom(ahcListener.getClass())) {
2850+
final byte[] bytesLocal = byteArrayOutputStream.toByteArray();
2851+
byteArrayOutputStream.reset();
2852+
WebSocketByteListener.class.cast(ahcListener).onMessage(bytesLocal);
2853+
}
2854+
}
2855+
}
2856+
} else {
2857+
if (WebSocketByteListener.class.isAssignableFrom(ahcListener.getClass())) {
2858+
WebSocketByteListener.class.cast(ahcListener).onFragment(bytes, last);
2859+
}
28062860
}
28072861
} catch (Throwable e) {
28082862
ahcListener.onError(e);

‎src/main/java/com/ning/http/client/providers/grizzly/GrizzlyAsyncHttpProviderConfig.java

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,13 @@ public static enum Property {
5959
MAX_HTTP_PACKET_HEADER_SIZE(Integer.class, HttpCodecFilter.DEFAULT_MAX_HTTP_PACKET_HEADER_SIZE),
6060

6161

62+
/**
63+
* By default, Websocket messages that are fragmented will be buffered. Once all
64+
* fragments have been accumulated, the appropriate onMessage() call back will be
65+
* invoked with the complete message. If this functionality is not desired, set
66+
* this property to false.
67+
*/
68+
BUFFER_WEBSOCKET_FRAGMENTS(Boolean.class, true)
6269

6370
;
6471

0 commit comments

Comments
 (0)