Introduce the capability to aggregate WebSocket continuation frames. #5357
Conversation
Motivation: Users may find it beneficial to aggregate WebSocket continuation frames and their text or binary frame into a single frame instead of managing them individually. Modifications: - Introduce properties `WebSocketServiceBuilder.aggregateContinuation()` and `WebSocketClientBuilder.aggregateContinuation()` to enable the aggregation of continuation frames. Result: - You now have the option to aggregate WebSocket continuation frames for more streamlined handling.
Codecov ReportAttention:
Additional details and impacted files@@ Coverage Diff @@
## main #5357 +/- ##
===========================================
+ Coverage 0 73.69% +73.69%
- Complexity 0 20126 +20126
===========================================
Files 0 1741 +1741
Lines 0 74392 +74392
Branches 0 9491 +9491
===========================================
+ Hits 0 54824 +54824
- Misses 0 15038 +15038
- Partials 0 4530 +4530 ☔ View full report in Codecov by Sentry. |
core/src/main/java/com/linecorp/armeria/internal/common/websocket/WebSocketFrameDecoder.java
Show resolved
Hide resolved
core/src/main/java/com/linecorp/armeria/internal/common/websocket/WebSocketFrameDecoder.java
Show resolved
Hide resolved
| // The message must not exceed 125 length: | ||
| // https://datatracker.ietf.org/doc/html/rfc6455/#section-5.5 |
There was a problem hiding this comment.
nit; this comment probably isn't related since it refers to control frames only
| // The message must not exceed 125 length: | |
| // https://datatracker.ietf.org/doc/html/rfc6455/#section-5.5 |
There was a problem hiding this comment.
The message is set to the reason phrase of the close frame and the close frame is one of control frames so I left the comment here. 😉
However, I've changed to logic to truncate the message if it exceeds the limit in WebSocketUtil so I've moved this comment to there. PTAL. 😉
| @@ -80,6 +81,7 @@ public final class WebSocketClientBuilder extends AbstractWebClientBuilder { | |||
| private int maxFramePayloadLength = DEFAULT_MAX_FRAME_PAYLOAD_LENGTH; | |||
| private boolean allowMaskMismatch; | |||
| private List<String> subprotocols = ImmutableList.of(); | |||
| private boolean aggregateContinuation; | |||
There was a problem hiding this comment.
minor question) Maybe it is just me, but I feel like most users won't be interested in continuation frames and true would be the more sensible default.
There was a problem hiding this comment.
I think we may provide a higher API that can automatically aggregate continuation frames with StreamMessage<String> or StreamMessage<(ByteBuf|byte[])>.
There was a problem hiding this comment.
most users won't be interested in continuation frame
I completely agree with your perspective on this matter, but I believe that setting true for aggregateContinuation as a default is a separate consideration.
If continuation frames are used that the server developers do not notice, the server may experience memory pressure. So I prefer to set it false. 🤔
There was a problem hiding this comment.
I think we may provide a higher API that can automatically aggregate continuation frames with StreamMessage or StreamMessage<(ByteBuf|byte[])>.
Could you share an example of the usage? I'm not sure how it's going to be used because what we need here is the stream message of WebSocketFrame instead of strings or byte arrays. 🤔
There was a problem hiding this comment.
Continuation is a low-level detail that many users might not be interested in for their business logic. I think we can abstract a set of WebSocket frames followed by continuation frames with StreamMessage so that users easily aggregate the whole message or subscribe to the messages one by one.
TextStreamMessage extends StreamMessage<String> {
CompletableFuture<String> aggregate();
}
ByteStreamMessage extends StreamMessage<byte[]> {
CompletableFuture<byte[]> aggregate();
}
interface WebSocketStreamHandler {
void onText(RequestContext ctx, TextStreamMessage texts, WebSocketWriter out);
void onBinary(RequestContext ctx, ByteStreamMessage bytes, WebSocketWriter out);
...
}class EchoWebSocketStreamHandler implements WebSocketStreamHandler {
void onText(RequestContext ctx, TextStreamMessage texts, WebSocketWriter out) {
// If users want to get the whole text
texts.aggregate().thenAccept(text -> {
out.write(text);
});
// To get a text frame one by one with backpressure
texts.subscribe(...)
}
void onBinary(RequestContext ctx, ByteStreamMessage bytes, WebSocketWriter out) {
...
}
}There was a problem hiding this comment.
I got it. I think I've seen a similar API from AKKA whether I like it or not. 😆
We can probably provide two different web socket handler APIs later.
The first one is the API you showed above and the second one is the simple callback style that you've used in your k8s PR.
https://github.com/line/armeria/pull/5167/files#diff-6dfa9f45d08fd2befd27967eec61221bcb1489bb0ceef6764d39f93aa0369d2bR138
For the second one, I think using this boolean value is more proper because the first one introduces another layer with a completable future that isn't needed.
There was a problem hiding this comment.
Agreed. We don't know which one is preferred by users. Both versions could be provided later if necessary.
core/src/main/java/com/linecorp/armeria/internal/common/websocket/WebSocketFrameDecoder.java
Show resolved
Hide resolved
Motivation:
Users may find it beneficial to aggregate WebSocket continuation frames and their text or binary frame into a single frame instead of managing them individually.
Modifications:
WebSocketServiceBuilder.aggregateContinuation()andWebSocketClientBuilder.aggregateContinuation()to enable the aggregation of continuation frames.Result: