Skip to content

Commit 3d2bb58

Browse files
slandelleslandelle
committed
Drop WebSocketXXXFragmentListener, close AsyncHttpClient#1231 
1 parent 9b1edab commit 3d2bb58

File tree

4 files changed

+37
-120
lines changed

4 files changed

+37
-120
lines changed

‎client/src/main/java/org/asynchttpclient/netty/handler/WebSocketHandler.java

Lines changed: 14 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@
1515

1616
import static io.netty.handler.codec.http.HttpResponseStatus.SWITCHING_PROTOCOLS;
1717
import static org.asynchttpclient.ws.WebSocketUtils.getAcceptKey;
18-
import io.netty.buffer.ByteBuf;
1918
import io.netty.channel.Channel;
2019
import io.netty.channel.ChannelHandler.Sharable;
2120
import io.netty.handler.codec.http.HttpHeaders;
@@ -33,7 +32,6 @@
3332

3433
import org.asynchttpclient.AsyncHandler.State;
3534
import org.asynchttpclient.AsyncHttpClientConfig;
36-
import org.asynchttpclient.HttpResponseBodyPart;
3735
import org.asynchttpclient.HttpResponseHeaders;
3836
import org.asynchttpclient.HttpResponseStatus;
3937
import org.asynchttpclient.netty.Callback;
@@ -77,7 +75,7 @@ public UpgradeCallback(NettyResponseFuture<?> future, Channel channel, HttpRespo
7775
private void invokeOnSucces(Channel channel, WebSocketUpgradeHandler h) {
7876
if (!h.touchSuccess()) {
7977
try {
80-
h.onSuccess(new NettyWebSocket(channel, responseHeaders.getHeaders(), config));
78+
h.onSuccess(new NettyWebSocket(channel, responseHeaders.getHeaders()));
8179
} catch (Exception ex) {
8280
logger.warn("onSuccess unexpected exception", ex);
8381
}
@@ -178,27 +176,23 @@ public void run() {
178176
}
179177

180178
private void handleFrame(Channel channel, WebSocketFrame frame, WebSocketUpgradeHandler handler, NettyWebSocket webSocket) throws Exception {
181-
if (frame instanceof CloseWebSocketFrame) {
179+
if (frame instanceof TextWebSocketFrame) {
180+
webSocket.onTextFragment((TextWebSocketFrame) frame);
181+
182+
} else if (frame instanceof BinaryWebSocketFrame) {
183+
webSocket.onBinaryFragment((BinaryWebSocketFrame) frame);
184+
185+
} else if (frame instanceof CloseWebSocketFrame) {
182186
Channels.setDiscard(channel);
183187
CloseWebSocketFrame closeFrame = (CloseWebSocketFrame) frame;
184188
webSocket.onClose(closeFrame.statusCode(), closeFrame.reasonText());
185189
Channels.silentlyCloseChannel(channel);
186-
} else {
187-
ByteBuf buf = frame.content();
188-
if (buf != null && buf.readableBytes() > 0) {
189-
HttpResponseBodyPart part = config.getResponseBodyPartFactory().newResponseBodyPart(buf, frame.isFinalFragment());
190-
handler.onBodyPartReceived(part);
191-
192-
if (frame instanceof BinaryWebSocketFrame) {
193-
webSocket.onBinaryFragment(part);
194-
} else if (frame instanceof TextWebSocketFrame) {
195-
webSocket.onTextFragment(part);
196-
} else if (frame instanceof PingWebSocketFrame) {
197-
webSocket.onPing(part);
198-
} else if (frame instanceof PongWebSocketFrame) {
199-
webSocket.onPong(part);
200-
}
201-
}
190+
191+
} else if (frame instanceof PingWebSocketFrame) {
192+
webSocket.onPing((PingWebSocketFrame) frame);
193+
194+
} else if (frame instanceof PongWebSocketFrame) {
195+
webSocket.onPong((PongWebSocketFrame) frame);
202196
}
203197
}
204198

‎client/src/main/java/org/asynchttpclient/netty/ws/NettyWebSocket.java

Lines changed: 15 additions & 100 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@
1414
package org.asynchttpclient.netty.ws;
1515

1616
import static io.netty.buffer.Unpooled.wrappedBuffer;
17-
import static java.nio.charset.StandardCharsets.UTF_8;
17+
import static org.asynchttpclient.util.ByteBufUtils.byteBuf2Bytes;
1818
import io.netty.channel.Channel;
1919
import io.netty.handler.codec.http.HttpHeaders;
2020
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
@@ -23,24 +23,16 @@
2323
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
2424
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
2525

26-
import java.io.ByteArrayOutputStream;
27-
import java.io.IOException;
2826
import java.net.SocketAddress;
29-
import java.util.ArrayList;
3027
import java.util.Collection;
31-
import java.util.List;
3228
import java.util.concurrent.ConcurrentLinkedQueue;
3329

34-
import org.asynchttpclient.AsyncHttpClientConfig;
35-
import org.asynchttpclient.HttpResponseBodyPart;
3630
import org.asynchttpclient.ws.WebSocket;
37-
import org.asynchttpclient.ws.WebSocketByteFragmentListener;
3831
import org.asynchttpclient.ws.WebSocketByteListener;
3932
import org.asynchttpclient.ws.WebSocketCloseCodeReasonListener;
4033
import org.asynchttpclient.ws.WebSocketListener;
4134
import org.asynchttpclient.ws.WebSocketPingListener;
4235
import org.asynchttpclient.ws.WebSocketPongListener;
43-
import org.asynchttpclient.ws.WebSocketTextFragmentListener;
4436
import org.asynchttpclient.ws.WebSocketTextListener;
4537
import org.slf4j.Logger;
4638
import org.slf4j.LoggerFactory;
@@ -52,21 +44,17 @@ public class NettyWebSocket implements WebSocket {
5244
protected final Channel channel;
5345
protected final HttpHeaders upgradeHeaders;
5446
protected final Collection<WebSocketListener> listeners;
55-
protected final int maxBufferSize;
56-
private int bufferSize;
57-
private List<byte[]> _fragments;
5847
private volatile boolean interestedInByteMessages;
5948
private volatile boolean interestedInTextMessages;
6049

61-
public NettyWebSocket(Channel channel, HttpHeaders upgradeHeaders, AsyncHttpClientConfig config) {
62-
this(channel, upgradeHeaders, config, new ConcurrentLinkedQueue<>());
50+
public NettyWebSocket(Channel channel, HttpHeaders upgradeHeaders) {
51+
this(channel, upgradeHeaders, new ConcurrentLinkedQueue<>());
6352
}
6453

65-
public NettyWebSocket(Channel channel, HttpHeaders upgradeHeaders, AsyncHttpClientConfig config, Collection<WebSocketListener> listeners) {
54+
public NettyWebSocket(Channel channel, HttpHeaders upgradeHeaders, Collection<WebSocketListener> listeners) {
6655
this.channel = channel;
6756
this.upgradeHeaders = upgradeHeaders;
6857
this.listeners = listeners;
69-
maxBufferSize = config.getWebSocketMaxBufferSize();
7058
}
7159

7260
@Override
@@ -207,118 +195,45 @@ public WebSocket removeWebSocketListener(WebSocketListener l) {
207195
return this;
208196
}
209197

210-
private List<byte[]> fragments() {
211-
if (_fragments == null)
212-
_fragments = new ArrayList<>(2);
213-
return _fragments;
214-
}
215-
216-
private void bufferFragment(byte[] buffer) {
217-
bufferSize += buffer.length;
218-
if (bufferSize > maxBufferSize) {
219-
onError(new Exception("Exceeded Netty Web Socket maximum buffer size of " + maxBufferSize));
220-
reset();
221-
close();
222-
} else {
223-
fragments().add(buffer);
224-
}
225-
}
226-
227-
private void reset() {
228-
fragments().clear();
229-
bufferSize = 0;
230-
}
231-
232198
private void notifyByteListeners(byte[] message) {
233199
for (WebSocketListener listener : listeners) {
234200
if (listener instanceof WebSocketByteListener)
235201
WebSocketByteListener.class.cast(listener).onMessage(message);
236202
}
237203
}
238204

239-
private void notifyTextListeners(byte[] bytes) {
240-
String message = new String(bytes, UTF_8);
205+
private void notifyTextListeners(String message) {
241206
for (WebSocketListener listener : listeners) {
242207
if (listener instanceof WebSocketTextListener)
243208
WebSocketTextListener.class.cast(listener).onMessage(message);
244209
}
245210
}
246211

247-
public void onBinaryFragment(HttpResponseBodyPart part) {
248-
249-
for (WebSocketListener listener : listeners) {
250-
if (listener instanceof WebSocketByteFragmentListener)
251-
WebSocketByteFragmentListener.class.cast(listener).onFragment(part);
252-
}
253-
212+
public void onBinaryFragment(BinaryWebSocketFrame frame) {
254213
if (interestedInByteMessages) {
255-
byte[] fragment = part.getBodyPartBytes();
256-
257-
if (part.isLast()) {
258-
if (bufferSize == 0) {
259-
notifyByteListeners(fragment);
260-
261-
} else {
262-
bufferFragment(fragment);
263-
notifyByteListeners(fragmentsBytes());
264-
}
265-
266-
reset();
267-
268-
} else
269-
bufferFragment(fragment);
214+
notifyByteListeners(byteBuf2Bytes(frame.content()));
270215
}
271216
}
272217

273-
private byte[] fragmentsBytes() {
274-
ByteArrayOutputStream os = new ByteArrayOutputStream(bufferSize);
275-
for (byte[] bytes : _fragments)
276-
try {
277-
os.write(bytes);
278-
} catch (IOException e) {
279-
// yeah, right
280-
}
281-
return os.toByteArray();
282-
}
283-
284-
public void onTextFragment(HttpResponseBodyPart part) {
285-
for (WebSocketListener listener : listeners) {
286-
if (listener instanceof WebSocketTextFragmentListener)
287-
WebSocketTextFragmentListener.class.cast(listener).onFragment(part);
288-
}
289-
218+
public void onTextFragment(TextWebSocketFrame frame) {
290219
if (interestedInTextMessages) {
291-
byte[] fragment = part.getBodyPartBytes();
292-
293-
if (part.isLast()) {
294-
if (bufferSize == 0) {
295-
notifyTextListeners(fragment);
296-
297-
} else {
298-
bufferFragment(fragment);
299-
notifyTextListeners(fragmentsBytes());
300-
}
301-
302-
reset();
303-
304-
} else
305-
bufferFragment(fragment);
220+
notifyTextListeners(frame.text());
306221
}
307222
}
308223

309-
public void onPing(HttpResponseBodyPart part) {
224+
public void onPing(PingWebSocketFrame frame) {
225+
byte[] bytes = byteBuf2Bytes(frame.content());
310226
for (WebSocketListener listener : listeners) {
311227
if (listener instanceof WebSocketPingListener)
312-
// bytes are cached in the part
313-
WebSocketPingListener.class.cast(listener).onPing(part.getBodyPartBytes());
228+
WebSocketPingListener.class.cast(listener).onPing(bytes);
314229
}
315230
}
316231

317-
public void onPong(HttpResponseBodyPart part) {
232+
public void onPong(PongWebSocketFrame frame) {
233+
byte[] bytes = byteBuf2Bytes(frame.content());
318234
for (WebSocketListener listener : listeners) {
319235
if (listener instanceof WebSocketPongListener)
320-
// bytes are cached in the part
321-
WebSocketPongListener.class.cast(listener).onPong(part.getBodyPartBytes());
236+
WebSocketPongListener.class.cast(listener).onPong(bytes);
322237
}
323238
}
324239
}

‎client/src/main/java/org/asynchttpclient/ws/WebSocketByteFragmentListener.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,11 @@
1717

1818
/**
1919
* Invoked when WebSocket binary fragments are received.
20+
*
21+
* Actually doesn't do anything, as chunks as assembled into full WebSocket frames.
22+
* Will be removed in 2.1.
2023
*/
24+
@Deprecated
2125
public interface WebSocketByteFragmentListener extends WebSocketListener {
2226

2327
/**

‎client/src/main/java/org/asynchttpclient/ws/WebSocketTextFragmentListener.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,11 @@
1717

1818
/**
1919
* Invoked when WebSocket text fragments are received.
20+
*
21+
* Actually doesn't do anything, as chunks as assembled into full WebSocket frames.
22+
* Will be removed in 2.1.
2023
*/
24+
@Deprecated
2125
public interface WebSocketTextFragmentListener extends WebSocketListener {
2226

2327
/**

0 commit comments

Comments
 (0)