feat: add msgpack support for WebSocket communication

- Implement format detection (`json`/`msgpack`) in WebSocket transport.
- Add `use_binary_protocol` option to enable `msgpack` encoding/decoding.
- Ensure compatibility with protocol parameters and set appropriate formats during connection.
- Add tests to verify `msgpack` and `json` behavior based on `use_binary_protocol` setting.

Author: ttypic

ably/realtime/connectionmanager.py

@@ -154,6 +154,9 @@ async def __get_transport_params(self) -> dict:
         params["v"] = protocol_version
         if self.connection_details:
             params["resume"] = self.connection_details.connection_key
+        # RTN2a: Set format to msgpack if use_binary_protocol is enabled
+        if self.options.use_binary_protocol:
+            params["format"] = "msgpack"
         return params
 
     async def close_impl(self) -> None:

ably/realtime/realtime_channel.py

@@ -558,7 +558,8 @@ def _on_message(self, proto_msg: dict) -> None:
         elif action == ProtocolMessageAction.MESSAGE:
             messages = []
             try:
-                messages = Message.from_encoded_array(proto_msg.get('messages'), context=self.__decoding_context)
+                messages = Message.from_encoded_array(proto_msg.get('messages'),
+                                                      cipher=self.cipher, context=self.__decoding_context)
                 self.__decoding_context.last_message_id = messages[-1].id
                 self.__channel_serial = channel_serial
             except AblyException as e:

ably/transport/websockettransport.py

@@ -8,6 +8,8 @@
 from enum import IntEnum
 from typing import TYPE_CHECKING
 
+import msgpack
+
 from ably.http.httputils import HttpUtils
 from ably.types.connectiondetails import ConnectionDetails
 from ably.util.eventemitter import EventEmitter
@@ -71,6 +73,7 @@ def __init__(self, connection_manager: ConnectionManager, host: str, params: dic
         self.is_disposed = False
         self.host = host
         self.params = params
+        self.format = params.get('format', 'json')
         super().__init__()
 
     def connect(self):
@@ -189,12 +192,18 @@ async def ws_read_loop(self):
             raise AblyException('ws_read_loop started with no websocket', 500, 50000)
         try:
             async for raw in self.websocket:
-                msg = json.loads(raw)
+                # Decode based on format
+                msg = self.decode_raw_websocket_frame(raw)
                 task = asyncio.create_task(self.on_protocol_message(msg))
                 task.add_done_callback(self.on_protcol_message_handled)
         except ConnectionClosedOK:
             return
 
+    def decode_raw_websocket_frame(self, raw: str | bytes) -> dict:
+        if self.format == 'msgpack':
+            return msgpack.unpackb(raw)
+        return json.loads(raw)
+
     def on_protcol_message_handled(self, task):
         try:
             exception = task.exception()
@@ -231,8 +240,13 @@ async def close(self):
     async def send(self, message: dict):
         if self.websocket is None:
             raise Exception()
-        raw_msg = json.dumps(message)
-        log.info(f'WebSocketTransport.send(): sending {raw_msg}')
+        # Encode based on format
+        if self.format == 'msgpack':
+            raw_msg = msgpack.packb(message)
+            log.info(f'WebSocketTransport.send(): sending msgpack message (length: {len(raw_msg)} bytes)')
+        else:
+            raw_msg = json.dumps(message)
+            log.info(f'WebSocketTransport.send(): sending {raw_msg}')
         await self.websocket.send(raw_msg)
 
     def set_idle_timer(self, timeout: float):

test/ably/realtime/realtimechannel_publish_test.py

@@ -3,9 +3,10 @@
 import pytest
 
 from ably.realtime.connection import ConnectionState
-from ably.realtime.realtime_channel import ChannelState
+from ably.realtime.realtime_channel import ChannelOptions, ChannelState
 from ably.transport.websockettransport import ProtocolMessageAction
 from ably.types.message import Message
+from ably.util.crypto import CipherParams
 from ably.util.exceptions import AblyException, IncompatibleClientIdException
 from test.ably.testapp import TestApp
 from test.ably.utils import BaseAsyncTestCase, WaitableEvent, assert_waiter
@@ -975,3 +976,37 @@ def on_message(message):
         assert data_received[1] == 'third message'
 
         await ably.close()
+
+    async def test_publish_with_encryption(self):
+        """Verify that encrypted messages can be published and received correctly"""
+        # Create connection with binary protocol enabled
+        ably = await TestApp.get_ably_realtime(use_binary_protocol=True)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
+
+        # Get channel with encryption enabled
+        cipher_params = CipherParams(secret_key=b'0123456789abcdef0123456789abcdef')
+        channel_options = ChannelOptions(cipher=cipher_params)
+        channel = ably.channels.get('encrypted_channel', channel_options)
+        await channel.attach()
+
+        received_data = None
+        data_received = WaitableEvent()
+        def on_message(message):
+            nonlocal received_data
+            try:
+                # message.decode()
+                received_data = message.data
+                data_received.finish()
+            except Exception as e:
+                data_received.finish()
+                raise e
+
+        await channel.subscribe(on_message)
+
+        await channel.publish('encrypted_event', 'sensitive data')
+
+        await data_received.wait()
+
+        assert received_data == 'sensitive data'
+
+        await ably.close()

test/ably/realtime/realtimeconnection_test.py

@@ -400,3 +400,63 @@ async def on_protocol_message(msg):
         await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
 
         await ably.close()
+
+    # RTN2f - Test msgpack format parameter when use_binary_protocol is enabled
+    async def test_connection_format_msgpack_with_binary_protocol(self):
+        """Test that format=msgpack is sent when use_binary_protocol=True"""
+        ably = await TestApp.get_ably_realtime(use_binary_protocol=True)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
+
+        received_raw_websocket_frames = []
+        transport = ably.connection.connection_manager.transport
+        original_decode_raw_websocket_frame = transport.decode_raw_websocket_frame
+
+        def intercepted_websocket_frame(data):
+            received_raw_websocket_frames.append(data)
+            return original_decode_raw_websocket_frame(data)
+
+        transport.decode_raw_websocket_frame = intercepted_websocket_frame
+
+        # Verify transport has format set to msgpack
+        assert ably.connection.connection_manager.transport is not None
+        assert ably.connection.connection_manager.transport.format == 'msgpack'
+
+        # Verify params include format=msgpack
+        assert ably.connection.connection_manager.transport.params.get('format') == 'msgpack'
+
+        await ably.channels.get('connection_test').publish('test', b'test')
+
+        assert len(received_raw_websocket_frames) > 0
+        assert all(isinstance(frame, bytes) for frame in received_raw_websocket_frames)
+
+        await ably.close()
+
+    async def test_connection_format_json_without_binary_protocol(self):
+        """Test that format defaults to json when use_binary_protocol=False"""
+        ably = await TestApp.get_ably_realtime(use_binary_protocol=False)
+        await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)
+
+        received_raw_websocket_frames = []
+        transport = ably.connection.connection_manager.transport
+        original_decode_raw_websocket_frame = transport.decode_raw_websocket_frame
+
+        def intercepted_websocket_frame(data):
+            received_raw_websocket_frames.append(data)
+            return original_decode_raw_websocket_frame(data)
+
+        transport.decode_raw_websocket_frame = intercepted_websocket_frame
+
+        # Verify transport has format set to json (default)
+        assert ably.connection.connection_manager.transport is not None
+        assert ably.connection.connection_manager.transport.format == 'json'
+
+        await ably.channels.get('connection_test').publish('test', b'test')
+
+        # Verify params don't include format parameter (or it's json)
+        transport_format = ably.connection.connection_manager.transport.params.get('format')
+        assert transport_format is None or transport_format == 'json'
+
+        assert len(received_raw_websocket_frames) > 0
+        assert all(isinstance(frame, str) for frame in received_raw_websocket_frames)
+
+        await ably.close()