This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-http.git
The following commit(s) were added to refs/heads/main by this push:
new 35174606d Add selective WebSocket message compression (#1172)
35174606d is described below
commit 35174606d8618071d5e0002c0cfe708c7baded3b
Author: Matthias Kurz <[email protected]>
AuthorDate: Fri Jul 24 00:04:44 2026 +0200
Add selective WebSocket message compression (#1172)
* Add selective WebSocket message compression #61
* Simplify selective WebSocket compression examples #61
* Order WebSocket server spec imports #61
---
.../handleWebSocketMessages.md | 3 +
.../handleWebSocketMessagesForOptionalProtocol.md | 3 +
.../handleWebSocketMessagesForProtocol.md | 3 +
.../main/paradox/server-side/websocket-support.md | 14 +
.../http/javadsl/server/WebSocketCoreExample.java | 12 +
.../scaladsl/server/WebSocketExampleSpec.scala | 14 +
.../pekko/http/impl/engine/ws/Handshake.scala | 57 +++-
.../impl/engine/ws/MessageToFrameRenderer.scala | 31 +-
.../http/impl/engine/ws/PerMessageDeflate.scala | 60 +++-
.../engine/ws/UpgradeToWebSocketLowLevel.scala | 12 +
.../pekko/http/impl/engine/ws/WebSocket.scala | 34 ++-
.../http/javadsl/model/ws/WebSocketUpgrade.scala | 67 +++++
.../http/scaladsl/model/ws/WebSocketUpgrade.scala | 107 +++++++
.../http/impl/engine/ws/WebSocketServerSpec.scala | 314 +++++++++++++++++++++
.../directives/WebSocketDirectivesSpec.scala | 83 +++++-
.../server/directives/WebSocketDirectives.scala | 44 ++-
.../server/directives/WebSocketDirectives.scala | 43 +++
17 files changed, 870 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 0d30e5a1c..fb675362f 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
@@ -13,6 +13,9 @@
The directive first checks if the request was a valid WebSocket handshake
request and if yes, it completes the request
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.
+
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 9c978800c..40f4222ed 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
@@ -18,6 +18,9 @@ If the `subprotocol` parameter is
@scala[None]@java[@javadoc:[empty](java.util.O
announced in the WebSocket request) @scala[contains `protocol`]@java[matches
the contained subprotocol]. If the client did not offer the protocol in question
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.
+
To support several subprotocols you may chain several
`handleWebSocketMessagesForOptionalProtocol` routes.
The `handleWebSocketMessagesForOptionalProtocol` directive is used as a
building block for @ref[WebSocket Directives](index.md) to handle websocket
messages.
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 9a3184a54..0664a1fba 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
@@ -18,6 +18,9 @@ The directive first checks if the request was a valid
WebSocket handshake reques
subprotocol name. If yes, the directive completes the request with the passed
handler. Otherwise, the request is
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.
+
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/docs/src/main/paradox/server-side/websocket-support.md
b/docs/src/main/paradox/server-side/websocket-support.md
index f0f8326a6..6a2256a68 100644
--- a/docs/src/main/paradox/server-side/websocket-support.md
+++ b/docs/src/main/paradox/server-side/websocket-support.md
@@ -171,6 +171,20 @@ The server exposes additional settings for the negotiated
extension under
If compression is enabled globally, a route can still decline compression for
a single accepted WebSocket by using the
`handleMessages` or `handleMessagesWith` overload with `compressionEnabled =
false`.
+After `permessage-deflate` is negotiated, an application can also select which
server-to-client messages are compressed.
+The compression filter is evaluated once for each outbound text or binary
message. Returning `true` compresses that
+message; returning `false` sends it uncompressed. The same decision applies to
every fragment of a streamed message.
+
+Scala
+: @@snip
[WebSocketExampleSpec.scala](/docs/src/test/scala/docs/http/scaladsl/server/WebSocketExampleSpec.scala)
{ #websocket-selective-compression }
+
+Java
+: @@snip
[WebSocketCoreExample.java](/docs/src/test/java/docs/http/javadsl/server/WebSocketCoreExample.java)
{ #websocket-selective-compression }
+
+The filter only controls outbound messages. It does not affect extension
negotiation, and the client can still send both
+compressed and uncompressed messages. The filter is not evaluated if
`permessage-deflate` was not negotiated. To disable
+compression in both directions for a security-sensitive endpoint, use the
`compressionEnabled = false` overload instead.
+
@@@ note
The `server_no_context_takeover` and `client_no_context_takeover` extension
parameters affect whether compression
dictionaries are retained across messages. Retaining context generally
improves compression ratio, while disabling
diff --git
a/docs/src/test/java/docs/http/javadsl/server/WebSocketCoreExample.java
b/docs/src/test/java/docs/http/javadsl/server/WebSocketCoreExample.java
index e28988938..b09f2ad79 100644
--- a/docs/src/test/java/docs/http/javadsl/server/WebSocketCoreExample.java
+++ b/docs/src/test/java/docs/http/javadsl/server/WebSocketCoreExample.java
@@ -21,6 +21,7 @@ import java.util.Optional;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
import org.apache.pekko.NotUsed;
import org.apache.pekko.actor.ActorSystem;
import org.apache.pekko.http.javadsl.ConnectionContext;
@@ -32,6 +33,7 @@ import org.apache.pekko.http.javadsl.model.StatusCodes;
import org.apache.pekko.http.javadsl.model.ws.Message;
import org.apache.pekko.http.javadsl.model.ws.TextMessage;
import org.apache.pekko.http.javadsl.model.ws.WebSocketRequest;
+import org.apache.pekko.http.javadsl.model.ws.WebSocketUpgrade;
import org.apache.pekko.http.javadsl.settings.ClientConnectionSettings;
import org.apache.pekko.http.javadsl.settings.ServerSettings;
import org.apache.pekko.http.javadsl.settings.WebSocketSettings;
@@ -123,6 +125,16 @@ public class WebSocketCoreExample {
// #websocket-handler
+ static HttpResponse selectiveCompression(
+ WebSocketUpgrade upgrade, Flow<Message, Message, NotUsed> handler) {
+ // #websocket-selective-compression
+ Predicate<Message> shouldCompress = Message::isText;
+
+ HttpResponse response = upgrade.handleMessagesWith(handler,
shouldCompress);
+ // #websocket-selective-compression
+ return response;
+ }
+
{
ActorSystem system = null;
Flow<HttpRequest, HttpResponse, NotUsed> handler = null;
diff --git
a/docs/src/test/scala/docs/http/scaladsl/server/WebSocketExampleSpec.scala
b/docs/src/test/scala/docs/http/scaladsl/server/WebSocketExampleSpec.scala
index a446a7522..9cb7b6390 100644
--- a/docs/src/test/scala/docs/http/scaladsl/server/WebSocketExampleSpec.scala
+++ b/docs/src/test/scala/docs/http/scaladsl/server/WebSocketExampleSpec.scala
@@ -125,6 +125,20 @@ class WebSocketExampleSpec extends AnyWordSpec with
Matchers with CompileOnlySpe
.onComplete(_ => system.terminate()) // and shutdown when done
}
+ "selective-compression-example" in compileOnlySpec {
+ import pekko.http.scaladsl.model.ws.{ Message, WebSocketUpgrade }
+ import pekko.stream.scaladsl.Flow
+
+ val upgrade: WebSocketUpgrade = null
+ val handler: Flow[Message, Message, Any] = null
+
+ // #websocket-selective-compression
+ val shouldCompress: Message => Boolean = _.isText
+
+ val response = upgrade.handleMessages(handler, shouldCompress)
+ // #websocket-selective-compression
+ }
+
"ping-server-example" in compileOnlySpec {
implicit val system: ActorSystem = null
val route = null
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/Handshake.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/Handshake.scala
index b9b293c9e..77a2096fe 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/Handshake.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/Handshake.scala
@@ -42,6 +42,9 @@ private[http] object Handshake {
object Server {
+ private val CompressEveryMessage: Message => Boolean = _ => true
+ private val CompressEveryFrame: FrameStart => Boolean = _ => true
+
/**
* Validates a client WebSocket handshake. Returns either
`OptionVal.Some(UpgradeToWebSocketLowLevel)` or
* `OptionVal.None`
@@ -140,12 +143,22 @@ private[http] object Handshake {
def handle(
handler: Either[Graph[FlowShape[FrameEvent, FrameEvent], Any],
Graph[FlowShape[Message, Message], Any]],
subprotocol: Option[String],
- compressionEnabled: Boolean): HttpResponse = {
+ compressionEnabled: Boolean,
+ shouldCompressMessage: Message => Boolean =
CompressEveryMessage,
+ shouldCompressFrame: FrameStart => Boolean =
CompressEveryFrame): HttpResponse = {
require(
subprotocol.forall(chosen =>
clientSupportedSubprotocols.contains(chosen)),
s"Tried to choose invalid subprotocol '$subprotocol' which
wasn't offered by the client: [${requestedProtocols.mkString(", ")}]")
val acceptedPerMessageDeflate = if (compressionEnabled)
perMessageDeflate else None
- buildResponse(key.get, handler, subprotocol,
acceptedPerMessageDeflate, settings, log)
+ buildResponse(
+ key.get,
+ handler,
+ subprotocol,
+ acceptedPerMessageDeflate,
+ settings,
+ log,
+ shouldCompressMessage,
+ shouldCompressFrame)
}
def handleFrames(
@@ -158,6 +171,13 @@ private[http] object Handshake {
compressionEnabled: Boolean): HttpResponse =
handle(Left(handlerFlow), subprotocol, compressionEnabled)
+ override private[http] def handleFrames(
+ handlerFlow: Graph[FlowShape[FrameEvent, FrameEvent], Any],
+ subprotocol: Option[String],
+ compressionEnabled: Boolean,
+ shouldCompress: FrameStart => Boolean): HttpResponse =
+ handle(Left(handlerFlow), subprotocol, compressionEnabled,
shouldCompressFrame = shouldCompress)
+
override def handleMessages(handlerFlow: Graph[FlowShape[Message,
Message], Any],
subprotocol: Option[String] = None): HttpResponse =
handle(Right(handlerFlow), subprotocol, compressionEnabled =
true)
@@ -167,6 +187,12 @@ private[http] object Handshake {
subprotocol: Option[String],
compressionEnabled: Boolean): HttpResponse =
handle(Right(handlerFlow), subprotocol, compressionEnabled)
+
+ override def handleMessages(
+ handlerFlow: Graph[FlowShape[Message, Message], Any],
+ subprotocol: Option[String],
+ shouldCompress: Message => Boolean): HttpResponse =
+ handle(Right(handlerFlow), subprotocol, compressionEnabled =
true, shouldCompressMessage = shouldCompress)
}
OptionVal.Some(header)
} else OptionVal.None
@@ -197,12 +223,33 @@ private[http] object Handshake {
subprotocol: Option[String],
perMessageDeflate: Option[PerMessageDeflate.Negotiated],
settings: WebSocketSettings,
- log: LoggingAdapter): HttpResponse = {
+ log: LoggingAdapter): HttpResponse =
+ buildResponse(
+ key,
+ handler,
+ subprotocol,
+ perMessageDeflate,
+ settings,
+ log,
+ CompressEveryMessage,
+ CompressEveryFrame)
+
+ private def buildResponse(
+ key: `Sec-WebSocket-Key`,
+ handler: Either[Graph[FlowShape[FrameEvent, FrameEvent], Any],
Graph[FlowShape[Message, Message], Any]],
+ subprotocol: Option[String],
+ perMessageDeflate: Option[PerMessageDeflate.Negotiated],
+ settings: WebSocketSettings,
+ log: LoggingAdapter,
+ shouldCompressMessage: Message => Boolean,
+ shouldCompressFrame: FrameStart => Boolean): HttpResponse = {
val frameHandler = handler match {
case Left(frameHandler) =>
-
perMessageDeflate.map(_.frameEventBidiFlow(settings.randomFactory).join(frameHandler)).getOrElse(frameHandler)
+ perMessageDeflate
+ .map(_.frameEventBidiFlow(settings.randomFactory,
shouldCompressFrame).join(frameHandler))
+ .getOrElse(frameHandler)
case Right(messageHandler) =>
- WebSocket.stack(serverSide = true, settings, perMessageDeflate =
perMessageDeflate, log = log)
+ WebSocket.stack(true, settings, perMessageDeflate, log,
shouldCompressMessage)
.join(messageHandler)
}
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/MessageToFrameRenderer.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/MessageToFrameRenderer.scala
index 1c5df2dce..0ae9389ce 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/MessageToFrameRenderer.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/MessageToFrameRenderer.scala
@@ -30,23 +30,34 @@ import pekko.http.scaladsl.model.ws._
*/
@InternalApi
private[http] object MessageToFrameRenderer {
- def create(serverSide: Boolean): Flow[Message, FrameStart, NotUsed] = {
- def strictFrames(opcode: Opcode, data: ByteString): Source[FrameStart, ?] =
+ def create(serverSide: Boolean): Flow[Message, FrameStart, NotUsed] =
+ create(serverSide, None)
+
+ def create(serverSide: Boolean, shouldCompress: Message => Boolean):
Flow[Message, FrameStart, NotUsed] =
+ create(serverSide, Some(shouldCompress))
+
+ private def create(
+ serverSide: Boolean,
+ shouldCompress: Option[Message => Boolean]): Flow[Message, FrameStart,
NotUsed] = {
+ def strictFrames(opcode: Opcode, data: ByteString, compress: Boolean):
Source[FrameStart, ?] =
// FIXME: fragment?
- Source.single(FrameEvent.fullFrame(opcode, None, data, fin = true))
+ Source.single(FrameEvent.fullFrame(opcode, None, data, fin = true, rsv1
= compress))
- def streamedFrames[M](opcode: Opcode, data: Source[ByteString, M]):
Source[FrameStart, Any] =
+ def streamedFrames[M](opcode: Opcode, data: Source[ByteString, M],
compress: Boolean): Source[FrameStart, Any] =
data.statefulMap(() => true)((isFirst, data) => {
val frameOpcode = if (isFirst) opcode else Opcode.Continuation
- (false, FrameEvent.fullFrame(frameOpcode, None, data, fin = false))
+ (false, FrameEvent.fullFrame(frameOpcode, None, data, fin = false,
rsv1 = isFirst && compress))
}, _ => None) ++ Source.single(FrameEvent.emptyLastContinuationFrame)
Flow[Message]
- .flatMapConcat {
- case BinaryMessage.Strict(data) => strictFrames(Opcode.Binary, data)
- case bm: BinaryMessage => streamedFrames(Opcode.Binary,
bm.dataStream)
- case TextMessage.Strict(text) => strictFrames(Opcode.Text,
ByteString(text, StandardCharsets.UTF_8))
- case tm: TextMessage => streamedFrames(Opcode.Text,
tm.textStream.via(Utf8Encoder))
+ .flatMapConcat { message =>
+ val compress = shouldCompress.exists(_(message))
+ message match {
+ case BinaryMessage.Strict(data) => strictFrames(Opcode.Binary, data,
compress)
+ case bm: BinaryMessage => streamedFrames(Opcode.Binary,
bm.dataStream, compress)
+ case TextMessage.Strict(text) => strictFrames(Opcode.Text,
ByteString(text, StandardCharsets.UTF_8), compress)
+ case tm: TextMessage => streamedFrames(Opcode.Text,
tm.textStream.via(Utf8Encoder), compress)
+ }
}
}
}
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
index 7bf669aa9..771da9027 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/PerMessageDeflate.scala
@@ -70,8 +70,16 @@ private[http] object PerMessageDeflate {
def bidiFlow: BidiFlow[FrameEventOrError, FrameEventOrError, FrameEvent,
FrameEvent, NotUsed] =
BidiFlow.fromFlows(inflaterFlow, deflaterFlow)
+ def messageBidiFlow: BidiFlow[FrameEventOrError, FrameEventOrError,
FrameEvent, FrameEvent, NotUsed] =
+ BidiFlow.fromFlows(inflaterFlow, selectiveDeflaterFlow(_.header.rsv1,
rsv1IndicatesCompression = true))
+
def frameEventBidiFlow(
maskRandom: () => Random): BidiFlow[FrameEvent, FrameEvent,
FrameEvent, FrameEvent, NotUsed] =
+ frameEventBidiFlow(maskRandom, _ => true)
+
+ def frameEventBidiFlow(
+ maskRandom: () => Random,
+ shouldCompress: FrameStart => Boolean): BidiFlow[FrameEvent,
FrameEvent, FrameEvent, FrameEvent, NotUsed] =
BidiFlow.fromFlows(
Flow[FrameEvent]
.via(Masking.unmaskIf(condition = true))
@@ -81,13 +89,26 @@ private[http] object PerMessageDeflate {
case FrameError(ex) => throw ex
}
.via(Masking.maskIf(condition = true, maskRandom)),
- deflaterFlow)
+ selectiveDeflaterFlow(shouldCompress, rsv1IndicatesCompression =
false))
private def inflaterFlow: Flow[FrameEventOrError, FrameEventOrError,
NotUsed] =
createInflaterFlow(clientNoContextTakeover, settings,
DefaultCompressionFactory)
private def deflaterFlow: Flow[FrameEvent, FrameEvent, NotUsed] =
createDeflaterFlow(serverNoContextTakeover, settings,
DefaultCompressionFactory)
+
+ private def selectiveDeflaterFlow(
+ shouldCompress: FrameStart => Boolean,
+ rsv1IndicatesCompression: Boolean): Flow[FrameEvent, FrameEvent,
NotUsed] =
+ Flow.fromGraph(new LifecycleMapConcatStage(
+ "PerMessageDeflate.deflater",
+ () =>
+ new DeflaterFlow(
+ serverNoContextTakeover,
+ settings,
+ DefaultCompressionFactory,
+ shouldCompress,
+ rsv1IndicatesCompression)))
}
private[ws] def createInflaterFlow(
@@ -104,7 +125,13 @@ private[http] object PerMessageDeflate {
compressionFactory: CompressionFactory): Flow[FrameEvent, FrameEvent,
NotUsed] =
Flow.fromGraph(new LifecycleMapConcatStage(
"PerMessageDeflate.deflater",
- () => new DeflaterFlow(noContextTakeover, settings, compressionFactory)))
+ () =>
+ new DeflaterFlow(
+ noContextTakeover,
+ settings,
+ compressionFactory,
+ _ => true,
+ rsv1IndicatesCompression = false)))
def negotiate(
requested: immutable.Seq[WebSocketExtension],
@@ -242,11 +269,14 @@ private[http] object PerMessageDeflate {
private final class DeflaterFlow(
noContextTakeover: Boolean,
settings: WebSocketCompressionSettingsImpl,
- compressionFactory: CompressionFactory)
+ compressionFactory: CompressionFactory,
+ shouldCompress: FrameStart => Boolean,
+ rsv1IndicatesCompression: Boolean)
extends LifecycleMapConcat[FrameEvent, FrameEvent] {
private var deflater =
compressionFactory.newDeflater(settings.compressionLevel)
private var frame: Option[UncompressedFrame] = None
private var messageInProgress = false
+ private var compressFragmentedMessage = false
private var bypassFrameInProgress = false
private val buffer = new Array[Byte](8192)
@@ -254,7 +284,7 @@ private[http] object PerMessageDeflate {
case FrameStart(header, _)
if (header.opcode == Protocol.Opcode.Text ||
header.opcode == Protocol.Opcode.Binary) &&
- (header.rsv1 || header.rsv2 || header.rsv3) =>
+ (header.rsv2 || header.rsv3 || (header.rsv1 &&
!rsv1IndicatesCompression)) =>
throw new ProtocolException("Unexpected reserved bit for outbound
WebSocket message")
case FrameStart(header, _)
if header.opcode == Protocol.Opcode.Continuation &&
@@ -265,18 +295,32 @@ private[http] object PerMessageDeflate {
header.opcode == Protocol.Opcode.Binary =>
if (messageInProgress || frame.isDefined)
throw new ProtocolException("Unexpected data frame while fragmented
message is open")
+ val compress = if (rsv1IndicatesCompression) header.rsv1 else
shouldCompress(start)
messageInProgress = !header.fin
- frame = Some(UncompressedFrame(header.copy(length = 0, rsv1 = true),
data, removeTail = header.fin))
- if (start.lastPart) finishFrame() else Nil
+ compressFragmentedMessage = compress && messageInProgress
+ if (compress) {
+ frame = Some(UncompressedFrame(header.copy(length = 0, rsv1 = true),
data, removeTail = header.fin))
+ if (start.lastPart) finishFrame() else Nil
+ } else {
+ bypassFrameInProgress = !start.lastPart
+ start :: Nil
+ }
case start @ FrameStart(header, _) if bypassFrameInProgress =>
throw new ProtocolException(s"Unexpected frame ${header.opcode} while
frame data is open")
case start @ FrameStart(header, _) if (frame.isDefined ||
messageInProgress) && header.opcode.isControl =>
bypassFrameInProgress = !start.lastPart
start :: Nil
case start @ FrameStart(header, data) if messageInProgress &&
header.opcode == Protocol.Opcode.Continuation =>
+ val compress = compressFragmentedMessage
messageInProgress = !header.fin
- frame = Some(UncompressedFrame(header.copy(length = 0), data,
removeTail = header.fin))
- if (start.lastPart) finishFrame() else Nil
+ if (!messageInProgress) compressFragmentedMessage = false
+ if (compress) {
+ frame = Some(UncompressedFrame(header.copy(length = 0), data,
removeTail = header.fin))
+ if (start.lastPart) finishFrame() else Nil
+ } else {
+ bypassFrameInProgress = !start.lastPart
+ start :: Nil
+ }
case start @ FrameStart(header, _) if frame.isDefined ||
messageInProgress =>
throw new ProtocolException(s"Unexpected frame ${header.opcode} while
fragmented message is open")
case data: FrameData if bypassFrameInProgress =>
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/UpgradeToWebSocketLowLevel.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/UpgradeToWebSocketLowLevel.scala
index ca5fbd165..d2274f4d2 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/UpgradeToWebSocketLowLevel.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/UpgradeToWebSocketLowLevel.scala
@@ -49,4 +49,16 @@ private[http] abstract class UpgradeToWebSocketLowLevel
extends InternalCustomHe
subprotocol: Option[String],
compressionEnabled: Boolean): HttpResponse =
handleFrames(handlerFlow, subprotocol)
+
+ /**
+ * The `shouldCompress` function is evaluated once for every outbound text
or binary message when
+ * `permessage-deflate` was negotiated. The decision for the initial frame
is retained for every continuation frame.
+ */
+ @InternalApi
+ private[http] def handleFrames(
+ handlerFlow: Graph[FlowShape[FrameEvent, FrameEvent], Any],
+ subprotocol: Option[String],
+ compressionEnabled: Boolean,
+ shouldCompress: FrameStart => Boolean): HttpResponse =
+ handleFrames(handlerFlow, subprotocol, compressionEnabled)
}
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/WebSocket.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/WebSocket.scala
index 5d1d37656..3e6fdcb7a 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/WebSocket.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/WebSocket.scala
@@ -50,12 +50,29 @@ private[http] object WebSocket {
perMessageDeflate: Option[PerMessageDeflate.Negotiated] = None,
closeTimeout: FiniteDuration = 3.seconds, // TODO put close timeout into
the settings?
log: LoggingAdapter): BidiFlow[FrameEvent, Message, Message, FrameEvent,
NotUsed] =
+ stack(serverSide, websocketSettings, perMessageDeflate, closeTimeout, log,
_ => true)
+
+ def stack(
+ serverSide: Boolean,
+ websocketSettings: WebSocketSettings,
+ perMessageDeflate: Option[PerMessageDeflate.Negotiated],
+ log: LoggingAdapter,
+ shouldCompress: Message => Boolean): BidiFlow[FrameEvent, Message,
Message, FrameEvent, NotUsed] =
+ stack(serverSide, websocketSettings, perMessageDeflate, 3.seconds, log,
shouldCompress)
+
+ def stack(
+ serverSide: Boolean,
+ websocketSettings: WebSocketSettings,
+ perMessageDeflate: Option[PerMessageDeflate.Negotiated],
+ closeTimeout: FiniteDuration,
+ log: LoggingAdapter,
+ shouldCompress: Message => Boolean): BidiFlow[FrameEvent, Message,
Message, FrameEvent, NotUsed] =
masking(serverSide, websocketSettings.randomFactory).atop(
FrameLogger.logFramesIfEnabled(websocketSettings.logFrames)).atop(
- perMessageDeflate.map(_.bidiFlow).getOrElse(BidiFlow.identity)).atop(
+
perMessageDeflate.map(_.messageBidiFlow).getOrElse(BidiFlow.identity)).atop(
frameHandling(serverSide, closeTimeout, log)).atop(
periodicKeepAlive(websocketSettings)).atop(
- messageAPI(serverSide, closeTimeout))
+ messageAPI(serverSide, closeTimeout, perMessageDeflate.map(_ =>
shouldCompress)))
/** The lowest layer that implements the binary protocol */
def framing: BidiFlow[ByteString, FrameEvent, FrameEvent, ByteString,
NotUsed] =
@@ -154,7 +171,14 @@ private[http] object WebSocket {
*/
def messageAPI(
serverSide: Boolean,
- closeTimeout: FiniteDuration): BidiFlow[FrameHandler.Output, Message,
Message, FrameOutHandler.Input, NotUsed] = {
+ closeTimeout: FiniteDuration): BidiFlow[FrameHandler.Output, Message,
Message, FrameOutHandler.Input, NotUsed] =
+ messageAPI(serverSide, closeTimeout, None)
+
+ private def messageAPI(
+ serverSide: Boolean,
+ closeTimeout: FiniteDuration,
+ shouldCompress: Option[Message => Boolean])
+ : BidiFlow[FrameHandler.Output, Message, Message, FrameOutHandler.Input,
NotUsed] = {
/* Collects user-level API messages from MessageDataParts */
val collectMessage: Flow[MessageDataPart, Message, NotUsed] =
Flow[MessageDataPart]
@@ -188,7 +212,9 @@ private[http] object WebSocket {
.named("ws-prepare-messages")
def renderMessages: Flow[Message, FrameStart, NotUsed] =
- MessageToFrameRenderer.create(serverSide)
+ shouldCompress
+ .map(MessageToFrameRenderer.create(serverSide, _))
+ .getOrElse(MessageToFrameRenderer.create(serverSide))
.named("ws-render-messages")
BidiFlow.fromGraph(GraphDSL.create() { implicit b =>
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 ceb70296a..04694d8df 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
@@ -14,6 +14,7 @@
package org.apache.pekko.http.javadsl.model.ws
import java.lang.{ Iterable => JIterable }
+import java.util.function.Predicate
import org.apache.pekko
import pekko.http.javadsl.model.HttpResponse
@@ -52,6 +53,20 @@ trait WebSocketUpgrade {
def handleMessagesWith(
handlerFlow: Graph[FlowShape[Message, Message], ? <: Any],
compressionEnabled: Boolean): HttpResponse
+ /**
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleMessagesWith(
+ handlerFlow: Graph[FlowShape[Message, Message], ? <: Any],
shouldCompress: Predicate[Message]): HttpResponse =
+ handleMessagesWith(handlerFlow)
+
/**
* 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.
The given subprotocol must be one
@@ -73,6 +88,23 @@ trait WebSocketUpgrade {
subprotocol: String,
compressionEnabled: Boolean): HttpResponse
+ /**
+ * 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.
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.
+ *
+ * @since 2.0.0
+ */
+ def handleMessagesWith(
+ handlerFlow: Graph[FlowShape[Message, Message], ? <: Any],
+ subprotocol: String,
+ shouldCompress: Predicate[Message]): HttpResponse =
+ handleMessagesWith(handlerFlow, subprotocol)
+
/**
* Returns a response that can be used to answer a WebSocket handshake
request. The connection will afterwards
* use the given inSink to handle WebSocket messages from the client and the
given outSource to send messages to the client.
@@ -91,6 +123,23 @@ trait WebSocketUpgrade {
def handleMessagesWith(inSink: Graph[SinkShape[Message], ? <: Any],
outSource: Graph[SourceShape[Message], ? <: Any],
compressionEnabled: Boolean): HttpResponse
+ /**
+ * Returns a response that can be used to answer a WebSocket handshake
request. The connection will afterwards
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleMessagesWith(
+ inSink: Graph[SinkShape[Message], ? <: Any],
+ outSource: Graph[SourceShape[Message], ? <: Any],
+ shouldCompress: Predicate[Message]): HttpResponse =
+ handleMessagesWith(inSink, outSource)
+
/**
* Returns a response that can be used to answer a WebSocket handshake
request. The connection will afterwards
* use the given inSink to handle WebSocket messages from the client and the
given outSource to send messages to the client.
@@ -111,4 +160,22 @@ trait WebSocketUpgrade {
*/
def handleMessagesWith(inSink: Graph[SinkShape[Message], ? <: Any],
outSource: Graph[SourceShape[Message], ? <: Any],
subprotocol: String, compressionEnabled: Boolean): HttpResponse
+
+ /**
+ * Returns a response that can be used to answer a WebSocket handshake
request. The connection will afterwards
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleMessagesWith(
+ inSink: Graph[SinkShape[Message], ? <: Any],
+ outSource: Graph[SourceShape[Message], ? <: Any],
+ subprotocol: String,
+ shouldCompress: Predicate[Message]): HttpResponse =
+ handleMessagesWith(inSink, outSource, subprotocol)
}
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 34de701c2..2f60e6c9f 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
@@ -14,6 +14,7 @@
package org.apache.pekko.http.scaladsl.model.ws
import java.lang.Iterable
+import java.util.function.{ Predicate => JPredicate }
import scala.collection.immutable
import org.apache.pekko
import pekko.NotUsed
@@ -71,6 +72,36 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade {
compressionEnabled: Boolean): HttpResponse =
handleMessages(handlerFlow, subprotocol)
+ /**
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleMessages(
+ handlerFlow: Graph[FlowShape[Message, Message], Any],
+ shouldCompress: Message => Boolean): HttpResponse =
+ handleMessages(handlerFlow, None, shouldCompress)
+
+ /**
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleMessages(
+ handlerFlow: Graph[FlowShape[Message, Message], Any],
+ subprotocol: Option[String],
+ shouldCompress: Message => Boolean): HttpResponse =
+ handleMessages(handlerFlow, subprotocol)
+
/**
* The high-level interface to create a WebSocket server based on "messages".
*
@@ -108,6 +139,38 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade {
compressionEnabled: Boolean): HttpResponse =
handleMessages(scaladsl.Flow.fromSinkAndSource(inSink, outSource),
subprotocol, compressionEnabled)
+ /**
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleMessagesWithSinkSource(
+ inSink: Graph[SinkShape[Message], Any],
+ outSource: Graph[SourceShape[Message], Any],
+ shouldCompress: Message => Boolean): HttpResponse =
+ handleMessages(scaladsl.Flow.fromSinkAndSource(inSink, outSource), None,
shouldCompress)
+
+ /**
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleMessagesWithSinkSource(
+ inSink: Graph[SinkShape[Message], Any],
+ outSource: Graph[SourceShape[Message], Any],
+ subprotocol: Option[String],
+ shouldCompress: Message => Boolean): HttpResponse =
+ handleMessages(scaladsl.Flow.fromSinkAndSource(inSink, outSource),
subprotocol, shouldCompress)
+
import scala.jdk.CollectionConverters._
/**
@@ -131,6 +194,16 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade {
compressionEnabled: Boolean): HttpResponse =
handleMessages(JavaMapping.toScala(handlerFlow), None, compressionEnabled)
+ /**
+ * Java API
+ *
+ * @since 2.0.0
+ */
+ override def handleMessagesWith(
+ handlerFlow: Graph[FlowShape[jm.ws.Message, jm.ws.Message], ? <: Any],
+ shouldCompress: JPredicate[jm.ws.Message]): HttpResponse =
+ handleMessages(JavaMapping.toScala(handlerFlow), message =>
shouldCompress.test(message))
+
/**
* Java API
*/
@@ -149,6 +222,17 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade {
compressionEnabled: Boolean): HttpResponse =
handleMessages(JavaMapping.toScala(handlerFlow), subprotocol =
Some(subprotocol), compressionEnabled)
+ /**
+ * Java API
+ *
+ * @since 2.0.0
+ */
+ override def handleMessagesWith(
+ handlerFlow: Graph[FlowShape[jm.ws.Message, jm.ws.Message], ? <: Any],
+ subprotocol: String,
+ shouldCompress: JPredicate[jm.ws.Message]): HttpResponse =
+ handleMessages(JavaMapping.toScala(handlerFlow), Some(subprotocol),
message => shouldCompress.test(message))
+
/**
* Java API
*/
@@ -167,6 +251,17 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade {
compressionEnabled: Boolean): HttpResponse =
handleMessages(createScalaFlow(inSink, outSource), None,
compressionEnabled)
+ /**
+ * Java API
+ *
+ * @since 2.0.0
+ */
+ override def handleMessagesWith(
+ inSink: Graph[SinkShape[jm.ws.Message], ? <: Any],
+ outSource: Graph[SourceShape[jm.ws.Message], ? <: Any],
+ shouldCompress: JPredicate[jm.ws.Message]): HttpResponse =
+ handleMessages(createScalaFlow(inSink, outSource), message =>
shouldCompress.test(message))
+
/**
* Java API
*
@@ -190,6 +285,18 @@ trait WebSocketUpgrade extends jm.ws.WebSocketUpgrade {
compressionEnabled: Boolean): HttpResponse =
handleMessages(createScalaFlow(inSink, outSource), subprotocol =
Some(subprotocol), compressionEnabled)
+ /**
+ * Java API
+ *
+ * @since 2.0.0
+ */
+ override def handleMessagesWith(
+ inSink: Graph[SinkShape[jm.ws.Message], ? <: Any],
+ outSource: Graph[SourceShape[jm.ws.Message], ? <: Any],
+ subprotocol: String,
+ shouldCompress: JPredicate[jm.ws.Message]): HttpResponse =
+ handleMessages(createScalaFlow(inSink, outSource), Some(subprotocol),
message => shouldCompress.test(message))
+
private def createScalaFlow(inSink: Graph[SinkShape[jm.ws.Message], ? <:
Any],
outSource: Graph[SourceShape[jm.ws.Message], ? <: Any]):
Graph[FlowShape[Message, Message], NotUsed] =
JavaMapping.toScala(scaladsl.Flow.fromSinkAndSourceMat(inSink,
outSource)(scaladsl.Keep.none): Graph[FlowShape[
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketServerSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketServerSpec.scala
index bffc62f9a..ec29e5c06 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketServerSpec.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketServerSpec.scala
@@ -17,11 +17,13 @@ import java.io.ByteArrayOutputStream
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
+import java.util.function.Predicate
import java.util.zip.Deflater
import java.util.zip.Inflater
import org.apache.pekko
import pekko.actor.ActorSystem
+import pekko.http.javadsl.model.ws.{ Message => JavaMessage, TextMessage =>
JavaTextMessage }
import pekko.http.scaladsl.model.ws._
import pekko.http.scaladsl.model.AttributeKeys.webSocketUpgrade
import pekko.stream.Materializer
@@ -536,6 +538,131 @@ class WebSocketServerSpec extends
PekkoSpecWithMaterializer("pekko.http.server.w
}
}
+ "select compression separately for each outbound message" in
Utils.assertAllStagesStopped {
+ new TestSetup {
+ sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
+
+ val request = expectRequest()
+ val upgrade = request.attribute(webSocketUpgrade)
+ val evaluated = new AtomicInteger()
+ val firstMessage = "compressed first server message"
+ val secondMessage = "plain"
+ val thirdMessage = "compressed third server message"
+ val response = upgrade.get.handleMessages(
+ Flow.fromSinkAndSource(
+ Sink.ignore,
+ Source(List(
+ TextMessage.Strict(firstMessage),
+ TextMessage.Strict(secondMessage),
+ TextMessage.Strict(thirdMessage)))),
+ None,
+ message => {
+ evaluated.incrementAndGet()
+
message.asInstanceOf[TextMessage.Strict].text.startsWith("compressed")
+ })
+ responses.sendNext(response)
+
+ expectResponseWithWipedDate(
+ """HTTP/1.1 101 Switching Protocols
+ |Upgrade: websocket
+ |Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
+ |Sec-WebSocket-Extensions: permessage-deflate
+ |Server: pekko-http/test
+ |Date: XXXX
+ |Connection: upgrade
+ |
+ |""")
+
+ val firstPayload = expectCompressedFrame(Protocol.Opcode.Text, fin =
true, rsv1 = true)
+ expectWSFrame(Protocol.Opcode.Text, ByteString(secondMessage), fin =
true)
+ val thirdPayload = expectCompressedFrame(Protocol.Opcode.Text, fin =
true, rsv1 = true)
+ inflatePerMessagesWithContext(firstPayload,
thirdPayload).map(_.utf8String) shouldEqual
+ Seq(firstMessage, thirdMessage)
+ evaluated.get() shouldEqual 3
+ expectWSCloseFrame(Protocol.CloseCodes.Regular)
+
+ sendWSCloseFrame(Protocol.CloseCodes.Regular, mask = true)
+ closeNetworkInput()
+ expectNetworkClose()
+ }
+ }
+
+ "select compression separately through the Java API" in
Utils.assertAllStagesStopped {
+ new TestSetup {
+ sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
+
+ val request = expectRequest()
+ val upgrade = request.attribute(webSocketUpgrade)
+ val response = upgrade.get.handleMessagesWith(
+ Flow.fromSinkAndSource(
+ Sink.foreach[JavaMessage](_ => ()),
+ Source(List[JavaMessage](
+ JavaTextMessage.create("plain"),
+ JavaTextMessage.create("compressed Java server message")))),
+ new Predicate[JavaMessage] {
+ override def test(message: JavaMessage): Boolean =
+ message.asTextMessage.getStrictText.startsWith("compressed")
+ })
+ responses.sendNext(response)
+
+ expectResponseWithWipedDate(
+ """HTTP/1.1 101 Switching Protocols
+ |Upgrade: websocket
+ |Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
+ |Sec-WebSocket-Extensions: permessage-deflate
+ |Server: pekko-http/test
+ |Date: XXXX
+ |Connection: upgrade
+ |
+ |""")
+
+ expectWSFrame(Protocol.Opcode.Text, ByteString("plain"), fin = true)
+ expectCompressedTextFrame("compressed Java server message")
+ expectWSCloseFrame(Protocol.CloseCodes.Regular)
+
+ sendWSCloseFrame(Protocol.CloseCodes.Regular, mask = true)
+ closeNetworkInput()
+ expectNetworkClose()
+ }
+ }
+
+ "not evaluate the outbound compression filter when permessage-deflate is
not negotiated" in
+ Utils.assertAllStagesStopped {
+ new TestSetup {
+ sendWebSocketRequest("")
+
+ val request = expectRequest()
+ val upgrade = request.attribute(webSocketUpgrade)
+ val evaluated = new AtomicInteger()
+ val response = upgrade.get.handleMessages(
+ Flow.fromSinkAndSource(Sink.ignore,
Source.single(TextMessage.Strict("plain server message"))),
+ None,
+ _ => {
+ evaluated.incrementAndGet()
+ true
+ })
+ responses.sendNext(response)
+
+ expectResponseWithWipedDate(
+ """HTTP/1.1 101 Switching Protocols
+ |Upgrade: websocket
+ |Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
+ |Server: pekko-http/test
+ |Date: XXXX
+ |Connection: upgrade
+ |
+ |""")
+
+ expectWSFrame(Protocol.Opcode.Text, ByteString("plain server
message"), fin = true)
+ evaluated.get() shouldEqual 0
+ expectWSCloseFrame(Protocol.CloseCodes.Regular)
+
+ sendWSCloseFrame(Protocol.CloseCodes.Regular, mask = true)
+ closeNetworkInput()
+ expectNetworkClose()
+ }
+ }
+
"deflate empty outbound messages sent by the application" in
Utils.assertAllStagesStopped {
new TestSetup {
sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
@@ -718,6 +845,48 @@ class WebSocketServerSpec extends
PekkoSpecWithMaterializer("pekko.http.server.w
tracking.awaitAllEnded()
}
+ "keep one uncompressed selection across a streamed outbound message" in
Utils.assertAllStagesStopped {
+ new TestSetup {
+ sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
+
+ val request = expectRequest()
+ val upgrade = request.attribute(webSocketUpgrade)
+ val evaluated = new AtomicInteger()
+ val response = upgrade.get.handleMessages(
+ Flow.fromSinkAndSource(
+ Sink.ignore,
+ Source.single(TextMessage(Source(List("streamed ", "plain ",
"message"))))),
+ None,
+ _ => {
+ evaluated.incrementAndGet()
+ false
+ })
+ responses.sendNext(response)
+
+ expectResponseWithWipedDate(
+ """HTTP/1.1 101 Switching Protocols
+ |Upgrade: websocket
+ |Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
+ |Sec-WebSocket-Extensions: permessage-deflate
+ |Server: pekko-http/test
+ |Date: XXXX
+ |Connection: upgrade
+ |
+ |""")
+
+ expectWSFrame(Protocol.Opcode.Text, ByteString("streamed "), fin =
false)
+ expectWSFrame(Protocol.Opcode.Continuation, ByteString("plain "),
fin = false)
+ expectWSFrame(Protocol.Opcode.Continuation, ByteString("message"),
fin = false)
+ expectWSFrame(Protocol.Opcode.Continuation, ByteString.empty, fin =
true)
+ evaluated.get() shouldEqual 1
+ expectWSCloseFrame(Protocol.CloseCodes.Regular)
+
+ sendWSCloseFrame(Protocol.CloseCodes.Regular, mask = true)
+ closeNetworkInput()
+ expectNetworkClose()
+ }
+ }
+
"fail invalid compressed messages with a protocol error" in
Utils.assertAllStagesStopped {
new TestSetup {
sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
@@ -1290,6 +1459,143 @@ class WebSocketServerSpec extends
PekkoSpecWithMaterializer("pekko.http.server.w
}
}
+ "select compression separately for low-level outbound messages" in
Utils.assertAllStagesStopped {
+ new TestSetup {
+ sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
+
+ val request = expectRequest()
+ val upgrade = request.attribute(webSocketUpgrade)
+ val evaluated = new AtomicInteger()
+ val firstMessage = ByteString("compressed first low-level message")
+ val secondMessage = ByteString("plain")
+ val thirdMessage = ByteString("compressed third low-level message")
+ val response =
upgrade.get.asInstanceOf[UpgradeToWebSocketLowLevel].handleFrames(
+ Flow.fromSinkAndSource(
+ Sink.ignore,
+ Source(List(
+ FrameEvent.fullFrame(Protocol.Opcode.Text, None, firstMessage,
fin = true),
+ FrameEvent.fullFrame(Protocol.Opcode.Text, None,
secondMessage, fin = true),
+ FrameEvent.fullFrame(Protocol.Opcode.Text, None, thirdMessage,
fin = true)))),
+ None,
+ compressionEnabled = true,
+ shouldCompress = start => {
+ evaluated.incrementAndGet()
+ start.data.utf8String.startsWith("compressed")
+ })
+ responses.sendNext(response)
+
+ expectResponseWithWipedDate(
+ """HTTP/1.1 101 Switching Protocols
+ |Upgrade: websocket
+ |Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
+ |Sec-WebSocket-Extensions: permessage-deflate
+ |Server: pekko-http/test
+ |Date: XXXX
+ |Connection: upgrade
+ |
+ |""")
+
+ val firstPayload = expectCompressedFrame(Protocol.Opcode.Text, fin =
true, rsv1 = true)
+ expectWSFrame(Protocol.Opcode.Text, secondMessage, fin = true)
+ val thirdPayload = expectCompressedFrame(Protocol.Opcode.Text, fin =
true, rsv1 = true)
+ inflatePerMessagesWithContext(firstPayload, thirdPayload)
shouldEqual Seq(firstMessage, thirdMessage)
+ evaluated.get() shouldEqual 3
+ netOut.expectComplete()
+
+ sendWSCloseFrame(Protocol.CloseCodes.Regular, mask = true)
+ closeNetworkInput()
+ }
+ }
+
+ "retain a low-level compression decision across continuation and control
frames" in
+ Utils.assertAllStagesStopped {
+ new TestSetup {
+ sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
+
+ val request = expectRequest()
+ val upgrade = request.attribute(webSocketUpgrade)
+ val evaluated = new AtomicInteger()
+ val response =
upgrade.get.asInstanceOf[UpgradeToWebSocketLowLevel].handleFrames(
+ Flow.fromSinkAndSource(
+ Sink.ignore,
+ Source(List(
+ FrameEvent.fullFrame(Protocol.Opcode.Text, None,
ByteString("plain "), fin = false),
+ FrameEvent.fullFrame(Protocol.Opcode.Ping, None,
ByteString("ping"), fin = true),
+ FrameEvent.fullFrame(Protocol.Opcode.Continuation, None,
ByteString("message"), fin = true)))),
+ None,
+ compressionEnabled = true,
+ shouldCompress = _ => {
+ evaluated.incrementAndGet()
+ false
+ })
+ responses.sendNext(response)
+
+ expectResponseWithWipedDate(
+ """HTTP/1.1 101 Switching Protocols
+ |Upgrade: websocket
+ |Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
+ |Sec-WebSocket-Extensions: permessage-deflate
+ |Server: pekko-http/test
+ |Date: XXXX
+ |Connection: upgrade
+ |
+ |""")
+
+ expectWSFrame(Protocol.Opcode.Text, ByteString("plain "), fin =
false)
+ expectWSFrame(Protocol.Opcode.Ping, ByteString("ping"), fin = true)
+ expectWSFrame(Protocol.Opcode.Continuation, ByteString("message"),
fin = true)
+ evaluated.get() shouldEqual 1
+ netOut.expectComplete()
+
+ sendWSCloseFrame(Protocol.CloseCodes.Regular, mask = true)
+ closeNetworkInput()
+ }
+ }
+
+ "pass every frame-data event through for an uncompressed low-level
message" in
+ Utils.assertAllStagesStopped {
+ new TestSetup {
+ sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
+
+ val request = expectRequest()
+ val upgrade = request.attribute(webSocketUpgrade)
+ val evaluated = new AtomicInteger()
+ val message = ByteString("plain split frame data")
+ val (firstPart, secondPart) = message.splitAt(8)
+ val response =
upgrade.get.asInstanceOf[UpgradeToWebSocketLowLevel].handleFrames(
+ Flow.fromSinkAndSource(
+ Sink.ignore,
+ Source(List(
+ FrameStart(FrameHeader(Protocol.Opcode.Text, None,
message.length, fin = true), firstPart),
+ FrameData(secondPart, lastPart = true)))),
+ None,
+ compressionEnabled = true,
+ shouldCompress = _ => {
+ evaluated.incrementAndGet()
+ false
+ })
+ responses.sendNext(response)
+
+ expectResponseWithWipedDate(
+ """HTTP/1.1 101 Switching Protocols
+ |Upgrade: websocket
+ |Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
+ |Sec-WebSocket-Extensions: permessage-deflate
+ |Server: pekko-http/test
+ |Date: XXXX
+ |Connection: upgrade
+ |
+ |""")
+
+ expectWSFrame(Protocol.Opcode.Text, message, fin = true)
+ evaluated.get() shouldEqual 1
+ netOut.expectComplete()
+
+ sendWSCloseFrame(Protocol.CloseCodes.Regular, mask = true)
+ closeNetworkInput()
+ }
+ }
+
"apply compression to low-level frame handlers" in
Utils.assertAllStagesStopped {
new TestSetup {
sendWebSocketRequest("Sec-WebSocket-Extensions:
permessage-deflate\r\n")
@@ -1493,6 +1799,14 @@ class WebSocketServerSpec extends
PekkoSpecWithMaterializer("pekko.http.server.w
}
}
+ private def inflatePerMessagesWithContext(messages: ByteString*):
Seq[ByteString] = {
+ val inflater = new Inflater(true)
+ try messages.map(message => inflateFrame(inflater, message ++
ByteString(0x00, 0x00, 0xFF.toByte, 0xFF.toByte)))
+ finally {
+ inflater.end()
+ }
+ }
+
private def inflateFrame(inflater: Inflater, data: ByteString): ByteString =
{
inflater.setInput(data.toArray)
val output = new ByteArrayOutputStream()
diff --git
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/directives/WebSocketDirectivesSpec.scala
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/directives/WebSocketDirectivesSpec.scala
index ad993b9ba..58f6c4da9 100644
---
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/directives/WebSocketDirectivesSpec.scala
+++
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/server/directives/WebSocketDirectivesSpec.scala
@@ -13,9 +13,15 @@
package org.apache.pekko.http.scaladsl.server.directives
+import java.util.function.Predicate
+
+import scala.collection.immutable
+
import org.apache.pekko
-import pekko.http.scaladsl.model.StatusCodes
-import pekko.http.scaladsl.model.headers.`Sec-WebSocket-Protocol`
+import pekko.http.impl.engine.server.InternalCustomHeader
+import pekko.http.scaladsl.model.AttributeKeys.webSocketUpgrade
+import pekko.http.scaladsl.model.{ HttpRequest, HttpResponse, StatusCodes }
+import pekko.http.scaladsl.model.headers.{ `Sec-WebSocket-Protocol`, Upgrade,
UpgradeProtocol }
import pekko.http.scaladsl.model.ws._
import pekko.http.scaladsl.server.{
ExpectedWebSocketRequestRejection,
@@ -24,7 +30,7 @@ import pekko.http.scaladsl.server.{
UnsupportedWebSocketSubprotocolRejection
}
import pekko.http.scaladsl.testkit.WSProbe
-import pekko.stream.OverflowStrategy
+import pekko.stream.{ FlowShape, Graph, OverflowStrategy }
import pekko.stream.scaladsl.{ Flow, Sink, Source }
import pekko.util.ByteString
@@ -49,6 +55,48 @@ class WebSocketDirectivesSpec extends RoutingSpec {
wsClient.expectCompletion()
}
}
+ "pass the outbound compression selector to the WebSocket upgrade" in {
+ var selectedSubprotocol = Option.empty[Option[String]]
+ var compressionDecision = Option.empty[Boolean]
+ val request = recordingWebSocketRequest() { (subprotocol,
shouldCompress) =>
+ selectedSubprotocol = Some(subprotocol)
+ compressionDecision = Some(shouldCompress(TextMessage("compress")))
+ }
+
+ request ~> handleWebSocketMessages(Flow[Message],
+ {
+ case TextMessage.Strict("compress") => true
+ case _ => false
+ }) ~> check {
+ isWebSocketUpgrade shouldEqual true
+ }
+
+ selectedSubprotocol shouldEqual Some(None)
+ compressionDecision shouldEqual Some(true)
+ }
+ "pass the Java outbound compression predicate and selected subprotocol to
the WebSocket upgrade" in {
+ var selectedSubprotocol = Option.empty[Option[String]]
+ var compressionDecision = Option.empty[Boolean]
+ val request = recordingWebSocketRequest("echo") { (subprotocol,
shouldCompress) =>
+ selectedSubprotocol = Some(subprotocol)
+ compressionDecision = Some(shouldCompress(TextMessage("compress")))
+ }
+ val shouldCompress = new Predicate[pekko.http.javadsl.model.ws.Message] {
+ override def test(message: pekko.http.javadsl.model.ws.Message):
Boolean = message.isText
+ }
+ val javaRoute =
pekko.http.javadsl.server.Directives.handleWebSocketMessagesForProtocol(
+ Flow[pekko.http.javadsl.model.ws.Message].asJava,
+ "echo",
+ shouldCompress)
+
+ request ~> javaRoute.asScala ~> check {
+ isWebSocketUpgrade shouldEqual true
+ header[`Sec-WebSocket-Protocol`].get.protocols shouldEqual
immutable.Seq("echo")
+ }
+
+ selectedSubprotocol shouldEqual Some(Some("echo"))
+ compressionDecision shouldEqual Some(true)
+ }
"choose subprotocol from offered ones" in {
val wsClient = WSProbe()
@@ -113,4 +161,33 @@ class WebSocketDirectivesSpec extends RoutingSpec {
def echo: Flow[Message, Message, Any] =
Flow[Message]
.buffer(1, OverflowStrategy.backpressure) // needed because a noop flow
hasn't any buffer that would start processing
+
+ private def recordingWebSocketRequest(offeredProtocols: String*)(
+ onHandle: (Option[String], Message => Boolean) => Unit): HttpRequest = {
+ val upgrade =
+ new InternalCustomHeader("UpgradeToWebSocketTestHeader") with
WebSocketUpgrade {
+ override def requestedProtocols: immutable.Seq[String] =
offeredProtocols.toList
+
+ override def handleMessages(
+ handlerFlow: Graph[FlowShape[Message, Message], Any],
+ subprotocol: Option[String]): HttpResponse =
+ throw new AssertionError("The selective-compression overload was not
called")
+
+ override def handleMessages(
+ handlerFlow: Graph[FlowShape[Message, Message], Any],
+ subprotocol: Option[String],
+ shouldCompress: Message => Boolean): HttpResponse = {
+ onHandle(subprotocol, shouldCompress)
+ HttpResponse(
+ StatusCodes.SwitchingProtocols,
+ headers =
+ Upgrade(UpgradeProtocol("websocket") :: Nil) ::
+ subprotocol.map(protocol => `Sec-WebSocket-Protocol`(protocol ::
Nil)).toList)
+ }
+ }
+
+ HttpRequest()
+ .addAttribute(webSocketUpgrade, upgrade)
+ .addHeader(upgrade)
+ }
}
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 71a0f8ecd..d4a1c9097 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
@@ -16,7 +16,7 @@ package directives
import java.util.{ List => JList }
import java.util.Optional
-import java.util.function.{ Function => JFunction }
+import java.util.function.{ Function => JFunction, Predicate }
import org.apache.pekko
import pekko.NotUsed
@@ -59,6 +59,17 @@ abstract class WebSocketDirectives extends
SecurityDirectives {
D.handleWebSocketMessages(adapt(handler))
}
+ /**
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleWebSocketMessages[T](handler: Flow[Message, Message, T],
shouldCompress: Predicate[Message]): Route =
+ RouteAdapter {
+ D.handleWebSocketMessages(adapt(handler), message =>
shouldCompress.test(message))
+ }
+
/**
* Handles WebSocket requests with the given handler if the given
subprotocol is offered in the request and
* rejects other requests with an [[ExpectedWebSocketRequestRejection]] or
an [[UnsupportedWebSocketSubprotocolRejection]].
@@ -68,6 +79,21 @@ abstract class WebSocketDirectives extends
SecurityDirectives {
D.handleWebSocketMessagesForProtocol(adapt(handler), subprotocol)
}
+ /**
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleWebSocketMessagesForProtocol[T](
+ handler: Flow[Message, Message, T],
+ subprotocol: String,
+ shouldCompress: Predicate[Message]): Route =
+ RouteAdapter {
+ D.handleWebSocketMessagesForProtocol(adapt(handler), subprotocol,
message => shouldCompress.test(message))
+ }
+
/**
* Handles WebSocket requests with the given handler and rejects other
requests with an
* [[ExpectedWebSocketRequestRejection]].
@@ -84,6 +110,22 @@ abstract class WebSocketDirectives extends
SecurityDirectives {
D.handleWebSocketMessagesForOptionalProtocol(adapt(handler),
subprotocol.asScala)
}
+ /**
+ * 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.
+ *
+ * @since 2.0.0
+ */
+ def handleWebSocketMessagesForOptionalProtocol[T](
+ handler: Flow[Message, Message, T],
+ subprotocol: Optional[String],
+ shouldCompress: Predicate[Message]): Route = RouteAdapter {
+ D.handleWebSocketMessagesForOptionalProtocol(
+ adapt(handler),
+ subprotocol.asScala,
+ message => shouldCompress.test(message))
+ }
+
private def adapt[T](handler: Flow[Message, Message, T]):
scaladsl.Flow[s.Message, s.Message, NotUsed] = {
scaladsl.Flow[s.Message].map(_.asJava).via(handler).map(_.asScala)
}
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 faff9f10e..5730ab568 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
@@ -60,6 +60,18 @@ trait WebSocketDirectives {
def handleWebSocketMessages(handler: Flow[Message, Message, Any]): Route =
handleWebSocketMessagesForOptionalProtocol(handler, None)
+ /**
+ * Handles WebSocket requests with the given handler and selectively
compresses outbound messages for which
+ * `shouldCompress` returns `true` when `permessage-deflate` was negotiated.
+ *
+ * @group websocket
+ * @since 2.0.0
+ */
+ def handleWebSocketMessages(
+ handler: Flow[Message, Message, Any],
+ shouldCompress: Message => Boolean): Route =
+ handleWebSocketMessagesForOptionalProtocol(handler, None, shouldCompress)
+
/**
* Handles WebSocket requests with the given handler if the given
subprotocol is offered in the request and
* rejects other requests with an [[ExpectedWebSocketRequestRejection]] or
an [[UnsupportedWebSocketSubprotocolRejection]].
@@ -69,6 +81,19 @@ trait WebSocketDirectives {
def handleWebSocketMessagesForProtocol(handler: Flow[Message, Message, Any],
subprotocol: String): Route =
handleWebSocketMessagesForOptionalProtocol(handler, Some(subprotocol))
+ /**
+ * 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.
+ *
+ * @group websocket
+ * @since 2.0.0
+ */
+ def handleWebSocketMessagesForProtocol(
+ handler: Flow[Message, Message, Any],
+ subprotocol: String,
+ shouldCompress: Message => Boolean): Route =
+ handleWebSocketMessagesForOptionalProtocol(handler, Some(subprotocol),
shouldCompress)
+
/**
* Handles WebSocket requests with the given handler and rejects other
requests with an
* [[ExpectedWebSocketRequestRejection]].
@@ -90,4 +115,22 @@ trait WebSocketDirectives {
else
reject(UnsupportedWebSocketSubprotocolRejection(subprotocol.get)) //
None.forall == true
}
+
+ /**
+ * Handles WebSocket requests with the given handler and selectively
compresses outbound messages for which
+ * `shouldCompress` returns `true` when `permessage-deflate` was negotiated.
+ *
+ * @group websocket
+ * @since 2.0.0
+ */
+ def handleWebSocketMessagesForOptionalProtocol(
+ handler: Flow[Message, Message, Any],
+ subprotocol: Option[String],
+ shouldCompress: Message => Boolean): Route =
+ extractWebSocketUpgrade { upgrade =>
+ if (subprotocol.forall(sub =>
upgrade.requestedProtocols.exists(_.equalsIgnoreCase(sub))))
+ complete(upgrade.handleMessages(handler, subprotocol, shouldCompress))
+ else
+ reject(UnsupportedWebSocketSubprotocolRejection(subprotocol.get)) //
None.forall == true
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
