From f8ab82badc94468bcee52948279ab91001a469ce Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Fri, 24 Jul 2026 00:04:14 +0200 Subject: [PATCH] Document selective WebSocket compression behavior #61 Clarify selector invocation, fragmentation, failure, and streaming semantics across the Scala and Java APIs and directive documentation. --- .../handleWebSocketMessages.md | 7 ++- ...dleWebSocketMessagesForOptionalProtocol.md | 7 ++- .../handleWebSocketMessagesForProtocol.md | 7 ++- .../javadsl/model/ws/WebSocketUpgrade.scala | 44 ++++++++++++++----- .../scaladsl/model/ws/WebSocketUpgrade.scala | 40 ++++++++++++----- .../directives/WebSocketDirectives.scala | 10 ++++- .../directives/WebSocketDirectives.scala | 15 +++++-- 7 files changed, 99 insertions(+), 31 deletions(-) diff --git a/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessages.md b/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessages.md index fb675362f..e153037b8 100644 --- a/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessages.md +++ b/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessages.md @@ -14,7 +14,12 @@ The directive first checks if the request was a valid WebSocket handshake reques with the passed handler. Otherwise, the request is rejected with an @apidoc[ExpectedWebSocketRequestRejection$]. The overload that accepts a `shouldCompress` filter can select compression separately for each outbound message after -`permessage-deflate` is negotiated. +`permessage-deflate` is negotiated. It is evaluated synchronously once for each outbound text or binary message, and +its result applies to every fragment of that message. The filter is not invoked for control frames or when compression +was not negotiated. It should be fast and non-blocking; throwing from it fails the WebSocket stream. + +For a streamed message, the complete payload and its final size are not available when the filter is evaluated. +Applications that select compression based on message size therefore need an explicit policy for streamed messages. WebSocket subprotocols offered in the `Sec-WebSocket-Protocol` header of the request are ignored. If you want to support several protocols use the @ref[handleWebSocketMessagesForProtocol](handleWebSocketMessagesForProtocol.md) directive, instead. diff --git a/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessagesForOptionalProtocol.md b/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessagesForOptionalProtocol.md index 40f4222ed..804f01399 100644 --- a/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessagesForOptionalProtocol.md +++ b/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessagesForOptionalProtocol.md @@ -19,7 +19,12 @@ announced in the WebSocket request) @scala[contains `protocol`]@java[matches the the request is rejected with an @apidoc[UnsupportedWebSocketSubprotocolRejection]. The overload that accepts a `shouldCompress` filter can select compression separately for each outbound message after -`permessage-deflate` is negotiated. +`permessage-deflate` is negotiated. It is evaluated synchronously once for each outbound text or binary message, and +its result applies to every fragment of that message. The filter is not invoked for control frames or when compression +was not negotiated. It should be fast and non-blocking; throwing from it fails the WebSocket stream. + +For a streamed message, the complete payload and its final size are not available when the filter is evaluated. +Applications that select compression based on message size therefore need an explicit policy for streamed messages. To support several subprotocols you may chain several `handleWebSocketMessagesForOptionalProtocol` routes. diff --git a/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessagesForProtocol.md b/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessagesForProtocol.md index 0664a1fba..bf04c7bea 100644 --- a/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessagesForProtocol.md +++ b/docs/src/main/paradox/routing-dsl/directives/websocket-directives/handleWebSocketMessagesForProtocol.md @@ -19,7 +19,12 @@ subprotocol name. If yes, the directive completes the request with the passed ha either rejected with an @apidoc[ExpectedWebSocketRequestRejection$] or an @apidoc[UnsupportedWebSocketSubprotocolRejection]. The overload that accepts a `shouldCompress` filter can select compression separately for each outbound message after -`permessage-deflate` is negotiated. +`permessage-deflate` is negotiated. It is evaluated synchronously once for each outbound text or binary message, and +its result applies to every fragment of that message. The filter is not invoked for control frames or when compression +was not negotiated. It should be fast and non-blocking; throwing from it fails the WebSocket stream. + +For a streamed message, the complete payload and its final size are not available when the filter is evaluated. +Applications that select compression based on message size therefore need an explicit policy for streamed messages. To support several subprotocols, for example at the same path, several instances of `handleWebSocketMessagesForProtocol` can be chained using `~` as you can see in the below example. diff --git a/http-core/src/main/scala/org/apache/pekko/http/javadsl/model/ws/WebSocketUpgrade.scala b/http-core/src/main/scala/org/apache/pekko/http/javadsl/model/ws/WebSocketUpgrade.scala index 04694d8df..c70b3edb1 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/javadsl/model/ws/WebSocketUpgrade.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/javadsl/model/ws/WebSocketUpgrade.scala @@ -57,9 +57,14 @@ trait WebSocketUpgrade { * Returns a response that can be used to answer a WebSocket handshake request. The connection will afterwards * use the given handlerFlow to handle WebSocket messages from the client. * - * If {@code permessage-deflate} is negotiated, {@code shouldCompress} is evaluated once for each outbound text or - * binary message. Returning {@code true} compresses that message and returning {@code false} sends it uncompressed. - * The filter does not affect inbound messages or whether compression is negotiated. + * If {@code permessage-deflate} is negotiated, {@code shouldCompress} is evaluated synchronously once for each + * outbound text or binary message. The result applies to every fragment of that message. Returning {@code true} + * compresses the message and returning {@code false} sends it uncompressed. The filter is not invoked for control + * frames or when compression was not negotiated, and it does not affect inbound messages or whether compression is + * negotiated. + * + * The filter should be fast and non-blocking; if it throws, the WebSocket stream fails. For streamed messages, the + * complete payload and its final size are not available when the filter is evaluated. * * @since 2.0.0 */ @@ -93,9 +98,14 @@ trait WebSocketUpgrade { * use the given handlerFlow to handle WebSocket messages from the client. The given subprotocol must be one * of the ones offered by the client. * - * If {@code permessage-deflate} is negotiated, {@code shouldCompress} is evaluated once for each outbound text or - * binary message. Returning {@code true} compresses that message and returning {@code false} sends it uncompressed. - * The filter does not affect inbound messages or whether compression is negotiated. + * If {@code permessage-deflate} is negotiated, {@code shouldCompress} is evaluated synchronously once for each + * outbound text or binary message. The result applies to every fragment of that message. Returning {@code true} + * compresses the message and returning {@code false} sends it uncompressed. The filter is not invoked for control + * frames or when compression was not negotiated, and it does not affect inbound messages or whether compression is + * negotiated. + * + * The filter should be fast and non-blocking; if it throws, the WebSocket stream fails. For streamed messages, the + * complete payload and its final size are not available when the filter is evaluated. * * @since 2.0.0 */ @@ -128,9 +138,14 @@ trait WebSocketUpgrade { * use the given inSink to handle WebSocket messages from the client and the given outSource to send messages to the * client. * - * If {@code permessage-deflate} is negotiated, {@code shouldCompress} is evaluated once for each outbound text or - * binary message. Returning {@code true} compresses that message and returning {@code false} sends it uncompressed. - * The filter does not affect inbound messages or whether compression is negotiated. + * If {@code permessage-deflate} is negotiated, {@code shouldCompress} is evaluated synchronously once for each + * outbound text or binary message. The result applies to every fragment of that message. Returning {@code true} + * compresses the message and returning {@code false} sends it uncompressed. The filter is not invoked for control + * frames or when compression was not negotiated, and it does not affect inbound messages or whether compression is + * negotiated. + * + * The filter should be fast and non-blocking; if it throws, the WebSocket stream fails. For streamed messages, the + * complete payload and its final size are not available when the filter is evaluated. * * @since 2.0.0 */ @@ -166,9 +181,14 @@ trait WebSocketUpgrade { * use the given inSink to handle WebSocket messages from the client and the given outSource to send messages to the * client. The given subprotocol must be one of the ones offered by the client. * - * If {@code permessage-deflate} is negotiated, {@code shouldCompress} is evaluated once for each outbound text or - * binary message. Returning {@code true} compresses that message and returning {@code false} sends it uncompressed. - * The filter does not affect inbound messages or whether compression is negotiated. + * If {@code permessage-deflate} is negotiated, {@code shouldCompress} is evaluated synchronously once for each + * outbound text or binary message. The result applies to every fragment of that message. Returning {@code true} + * compresses the message and returning {@code false} sends it uncompressed. The filter is not invoked for control + * frames or when compression was not negotiated, and it does not affect inbound messages or whether compression is + * negotiated. + * + * The filter should be fast and non-blocking; if it throws, the WebSocket stream fails. For streamed messages, the + * complete payload and its final size are not available when the filter is evaluated. * * @since 2.0.0 */ diff --git a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/model/ws/WebSocketUpgrade.scala b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/model/ws/WebSocketUpgrade.scala index 2f60e6c9f..779e40d1d 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/model/ws/WebSocketUpgrade.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/model/ws/WebSocketUpgrade.scala @@ -75,9 +75,13 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade { /** * The high-level interface to create a WebSocket server based on "messages". * - * If `permessage-deflate` is negotiated, `shouldCompress` is evaluated once for each outbound text or binary - * message. Returning `true` compresses that message and returning `false` sends it uncompressed. The filter does not - * affect inbound messages or whether compression is negotiated. + * If `permessage-deflate` is negotiated, `shouldCompress` is evaluated synchronously once for each outbound text or + * binary message. The result applies to every fragment of that message. Returning `true` compresses the message and + * returning `false` sends it uncompressed. The filter is not invoked for control frames or when compression was not + * negotiated, and it does not affect inbound messages or whether compression is negotiated. + * + * The filter should be fast and non-blocking; if it throws, the WebSocket stream fails. For streamed messages, the + * complete payload and its final size are not available when the filter is evaluated. * * @since 2.0.0 */ @@ -90,9 +94,13 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade { * The high-level interface to create a WebSocket server based on "messages". * * Optionally, a subprotocol out of the ones requested by the client can be chosen. If `permessage-deflate` is - * negotiated, `shouldCompress` is evaluated once for each outbound text or binary message. Returning `true` - * compresses that message and returning `false` sends it uncompressed. The filter does not affect inbound messages - * or whether compression is negotiated. + * negotiated, `shouldCompress` is evaluated synchronously once for each outbound text or binary message. The result + * applies to every fragment of that message. Returning `true` compresses the message and returning `false` sends it + * uncompressed. The filter is not invoked for control frames or when compression was not negotiated, and it does not + * affect inbound messages or whether compression is negotiated. + * + * The filter should be fast and non-blocking; if it throws, the WebSocket stream fails. For streamed messages, the + * complete payload and its final size are not available when the filter is evaluated. * * @since 2.0.0 */ @@ -142,9 +150,13 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade { /** * The high-level interface to create a WebSocket server based on "messages". * - * If `permessage-deflate` is negotiated, `shouldCompress` is evaluated once for each outbound text or binary - * message. Returning `true` compresses that message and returning `false` sends it uncompressed. The filter does not - * affect inbound messages or whether compression is negotiated. + * If `permessage-deflate` is negotiated, `shouldCompress` is evaluated synchronously once for each outbound text or + * binary message. The result applies to every fragment of that message. Returning `true` compresses the message and + * returning `false` sends it uncompressed. The filter is not invoked for control frames or when compression was not + * negotiated, and it does not affect inbound messages or whether compression is negotiated. + * + * The filter should be fast and non-blocking; if it throws, the WebSocket stream fails. For streamed messages, the + * complete payload and its final size are not available when the filter is evaluated. * * @since 2.0.0 */ @@ -158,9 +170,13 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade { * The high-level interface to create a WebSocket server based on "messages". * * Optionally, a subprotocol out of the ones requested by the client can be chosen. If `permessage-deflate` is - * negotiated, `shouldCompress` is evaluated once for each outbound text or binary message. Returning `true` - * compresses that message and returning `false` sends it uncompressed. The filter does not affect inbound messages - * or whether compression is negotiated. + * negotiated, `shouldCompress` is evaluated synchronously once for each outbound text or binary message. The result + * applies to every fragment of that message. Returning `true` compresses the message and returning `false` sends it + * uncompressed. The filter is not invoked for control frames or when compression was not negotiated, and it does not + * affect inbound messages or whether compression is negotiated. + * + * The filter should be fast and non-blocking; if it throws, the WebSocket stream fails. For streamed messages, the + * complete payload and its final size are not available when the filter is evaluated. * * @since 2.0.0 */ diff --git a/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/WebSocketDirectives.scala b/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/WebSocketDirectives.scala index d4a1c9097..7c568dc74 100644 --- a/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/WebSocketDirectives.scala +++ b/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/WebSocketDirectives.scala @@ -62,6 +62,9 @@ abstract class WebSocketDirectives extends SecurityDirectives { /** * Handles WebSocket requests with the given handler and selectively compresses outbound messages for which * {@code shouldCompress} returns {@code true} when {@code permessage-deflate} was negotiated. + * The filter is evaluated synchronously once per outbound text or binary message, and the result applies to all + * fragments. It should be fast and non-blocking; if it throws, the WebSocket stream fails. A streamed message's + * complete payload and final size are not available when the filter is evaluated. * * @since 2.0.0 */ @@ -82,7 +85,9 @@ abstract class WebSocketDirectives extends SecurityDirectives { /** * Handles WebSocket requests with the given handler if the given subprotocol is offered and selectively compresses * outbound messages for which {@code shouldCompress} returns {@code true} when {@code permessage-deflate} was - * negotiated. + * negotiated. The filter is evaluated synchronously once per outbound text or binary message, and the result applies + * to all fragments. It should be fast and non-blocking; if it throws, the WebSocket stream fails. A streamed + * message's complete payload and final size are not available when the filter is evaluated. * * @since 2.0.0 */ @@ -113,6 +118,9 @@ abstract class WebSocketDirectives extends SecurityDirectives { /** * Handles WebSocket requests with the given handler and selectively compresses outbound messages for which * {@code shouldCompress} returns {@code true} when {@code permessage-deflate} was negotiated. + * The filter is evaluated synchronously once per outbound text or binary message, and the result applies to all + * fragments. It should be fast and non-blocking; if it throws, the WebSocket stream fails. A streamed message's + * complete payload and final size are not available when the filter is evaluated. * * @since 2.0.0 */ diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/server/directives/WebSocketDirectives.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/server/directives/WebSocketDirectives.scala index 5730ab568..6ebb88506 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/server/directives/WebSocketDirectives.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/server/directives/WebSocketDirectives.scala @@ -62,7 +62,10 @@ trait WebSocketDirectives { /** * Handles WebSocket requests with the given handler and selectively compresses outbound messages for which - * `shouldCompress` returns `true` when `permessage-deflate` was negotiated. + * `shouldCompress` returns `true` when `permessage-deflate` was negotiated. The filter is evaluated synchronously + * once per outbound text or binary message, and the result applies to all fragments. It should be fast and + * non-blocking; if it throws, the WebSocket stream fails. A streamed message's complete payload and final size are + * not available when the filter is evaluated. * * @group websocket * @since 2.0.0 @@ -83,7 +86,10 @@ trait WebSocketDirectives { /** * Handles WebSocket requests with the given handler if the given subprotocol is offered and selectively compresses - * outbound messages for which `shouldCompress` returns `true` when `permessage-deflate` was negotiated. + * outbound messages for which `shouldCompress` returns `true` when `permessage-deflate` was negotiated. The filter + * is evaluated synchronously once per outbound text or binary message, and the result applies to all fragments. It + * should be fast and non-blocking; if it throws, the WebSocket stream fails. A streamed message's complete payload + * and final size are not available when the filter is evaluated. * * @group websocket * @since 2.0.0 @@ -118,7 +124,10 @@ trait WebSocketDirectives { /** * Handles WebSocket requests with the given handler and selectively compresses outbound messages for which - * `shouldCompress` returns `true` when `permessage-deflate` was negotiated. + * `shouldCompress` returns `true` when `permessage-deflate` was negotiated. The filter is evaluated synchronously + * once per outbound text or binary message, and the result applies to all fragments. It should be fast and + * non-blocking; if it throws, the WebSocket stream fails. A streamed message's complete payload and final size are + * not available when the filter is evaluated. * * @group websocket * @since 2.0.0