Add selective WebSocket message compression

[mkurz]

Summary

Adds server-side per-message compression selection when permessage-deflate has been negotiated.

Applications can now decide independently for every outbound text or binary message whether it should be compressed. This supports policies such as skipping sensitive, already-compressed, or small messages while retaining compression for the rest of the WebSocket connection.

API

The Scala API accepts a Message => Boolean selector, while the Java API accepts a Predicate<Message>.

Selector overloads are available for:

• WebSocketUpgrade.handleMessages
• WebSocketUpgrade.handleMessagesWithSinkSource
• handleWebSocketMessages
• handleWebSocketMessagesForProtocol
• handleWebSocketMessagesForOptionalProtocol

The internal low-level frame API additionally accepts a FrameStart => Boolean selector.

Behavior

• The selector is evaluated exactly once for each outbound text or binary message.
• It is evaluated only when permessage-deflate was negotiated.
• Control frames are never passed to the selector or compressed.
• One decision is retained across all fragments of a streamed or fragmented message, including when control frames are interleaved.
• Sending an uncompressed message does not modify the DEFLATE context, so later compressed messages continue with the correct context-takeover state.
• Existing APIs retain their current behavior and compress every outbound data message after successful negotiation.

Returning true selects compression for the message. It does not enable or negotiate compression by itself.

Implementation

The high-level message API evaluates the selector before rendering and carries the result on the initial frame for the compression stage.

The low-level frame API evaluates its selector directly for the initial text or binary frame. The compression stage retains that choice until the final continuation frame.

This is similar to Netty's per-message WebSocket extension filter while using Pekko HTTP's existing streaming frame pipeline.

Compatibility

All existing overloads and internal entry points remain available and delegate to the previous compress-all behavior.

The new Scala trait methods are concrete delegating methods, so existing implementations do not need to implement them.

Testing

Coverage includes:

• Alternating compressed and uncompressed messages
• Scala and Java selectors
• Strict and streamed messages
• Fragmented messages with interleaved control frames
• Context takeover across skipped messages
• Selector evaluation count
• No selector evaluation without negotiation
• High-level and low-level frame APIs
• Scala and Java routing-directive delegation
• Subprotocol preservation
• Binary compatibility on Scala 2.13 and Scala 3

Validation performed with:

sbt applyCodeStyle
sbt +mimaReportBinaryIssues
sbt '+http-tests/testOnly org.apache.pekko.http.scaladsl.server.directives.WebSocketDirectivesSpec'
sbt 'http-core/testOnly org.apache.pekko.http.impl.engine.ws.WebSocketServerSpec'

References

• Please add support for WebSocket compression #61
• Add server-side WebSocket compression #1114

[mkurz]

This is needed for Play so we can implement a threshold:

• Add a WebSocket compression threshold playframework/playframework#14120

[mkurz]

any chance this gets merged within the next 5 hours so it makes it into the next nightly (so tomorrow I can build against that in Play)? (no pressure however, just checking)

[pjfanning]

lgtm

[pjfanning]

@mkurz could you fix the merge conflicts and I'll merge.
If anyone spots any issues post merge, we can look at fixing them.
It will be a few weeks before we get around to a 2.0.0-M2 RC because I want to get the Pekko Core 2.0.0-M4 out first.

[mkurz]

Thanks!

[pjfanning]

2.0.0-M1+205-35174606-SNAPSHOT is published