Add serialize and deserialize method to RedisPubSubChannelLayer (#281)

* Add serializer and deserialize method to RedisPubSubChannelLayer

* move serialization methods directly to RedisPubSubChannelLayer

* Pass a full reference of RedisPubSubChannelLayer to RedisPubSubLoopLayer instance

* Fix linter errors

* Fix linter errors (again)

Using the version of `black` to match the repo this time.

* Not specifying these parameters (since they are the defaults)

There are many _other_ parameters which are not specified. It's more clear this way to show we are using _all_ defaults.

Co-authored-by: Jakub Stawowy <[email protected]>
Co-authored-by: Ryan Henning <[email protected]>

Author: Kub-AT

channels_redis/pubsub.py

@@ -54,13 +54,29 @@ def __getattr__(self, name):
         else:
             return getattr(self._get_layer(), name)
 
+    def serialize(self, message):
+        """
+        Serializes message to a byte string.
+        """
+        return msgpack.packb(message)
+
+    def deserialize(self, message):
+        """
+        Deserializes from a byte string.
+        """
+        return msgpack.unpackb(message)
+
     def _get_layer(self):
         loop = get_running_loop()
 
         try:
             layer = self._layers[loop]
         except KeyError:
-            layer = RedisPubSubLoopLayer(*self._args, **self._kwargs)
+            layer = RedisPubSubLoopLayer(
+                *self._args,
+                **self._kwargs,
+                channel_layer=self,
+            )
             self._layers[loop] = layer
             _wrap_close(self, loop)
 
@@ -80,7 +96,13 @@ class RedisPubSubLoopLayer:
     """
 
     def __init__(
-        self, hosts=None, prefix="asgi", on_disconnect=None, on_reconnect=None, **kwargs
+        self,
+        hosts=None,
+        prefix="asgi",
+        on_disconnect=None,
+        on_reconnect=None,
+        channel_layer=None,
+        **kwargs,
     ):
         if hosts is None:
             hosts = [("localhost", 6379)]
@@ -92,6 +114,7 @@ def __init__(
 
         self.on_disconnect = on_disconnect
         self.on_reconnect = on_reconnect
+        self.channel_layer = channel_layer
 
         # Each consumer gets its own *specific* channel, created with the `new_channel()` method.
         # This dict maps `channel_name` to a queue of messages for that channel.
@@ -133,7 +156,7 @@ async def send(self, channel, message):
         Send a message onto a (general or specific) channel.
         """
         shard = self._get_shard(channel)
-        await shard.publish(channel, message)
+        await shard.publish(channel, self.channel_layer.serialize(message))
 
     async def new_channel(self, prefix="specific."):
         """
@@ -180,7 +203,7 @@ async def receive(self, channel):
                     # We don't re-raise here because we want the CancelledError to be the one re-raised.
             raise
 
-        return msgpack.unpackb(message)
+        return self.channel_layer.deserialize(message)
 
     ################################################################################
     # Groups extension
@@ -225,7 +248,7 @@ async def group_send(self, group, message):
         """
         group_channel = self._get_group_channel_name(group)
         shard = self._get_shard(group_channel)
-        await shard.publish(group_channel, message)
+        await shard.publish(group_channel, self.channel_layer.serialize(message))
 
     ################################################################################
     # Flush extension
@@ -271,7 +294,7 @@ def __init__(self, host, channel_layer):
 
     async def publish(self, channel, message):
         conn = await self._get_pub_conn()
-        await conn.publish(channel, msgpack.packb(message))
+        await conn.publish(channel, message)
 
     async def subscribe(self, channel):
         if channel not in self._subscribed_to:
@@ -397,7 +420,9 @@ async def _do_receiving(self):
     def _notify_consumers(self, mtype):
         if mtype is not None:
             for channel in self.channel_layer.channels.values():
-                channel.put_nowait(msgpack.packb({"type": mtype}))
+                channel.put_nowait(
+                    self.channel_layer.channel_layer.serialize({"type": mtype})
+                )
 
     async def _ensure_redis(self):
         if self._redis is None:

tests/test_core.py

@@ -641,3 +641,26 @@ def test_receive_buffer_respects_capacity():
     assert buff.qsize() == capacity
     messages = [buff.get_nowait() for _ in range(capacity)]
     assert list(range(9900, 10000)) == messages
+
+
+def test_serialize():
+    """
+    Test default serialization method
+    """
+    message = {"a": True, "b": None, "c": {"d": []}}
+    channel_layer = RedisChannelLayer()
+    serialized = channel_layer.serialize(message)
+    assert isinstance(serialized, bytes)
+    assert serialized[12:] == b"\x83\xa1a\xc3\xa1b\xc0\xa1c\x81\xa1d\x90"
+
+
+def test_deserialize():
+    """
+    Test default deserialization method
+    """
+    message = b"Q\x0c\xbb?Q\xbc\xe3|D\xfd9\x00\x83\xa1a\xc3\xa1b\xc0\xa1c\x81\xa1d\x90"
+    channel_layer = RedisChannelLayer()
+    deserialized = channel_layer.deserialize(message)
+
+    assert isinstance(deserialized, dict)
+    assert deserialized == {"a": True, "b": None, "c": {"d": []}}

tests/test_pubsub.py

@@ -129,6 +129,34 @@ async def test_random_reset__channel_name(channel_layer):
     assert channel_name_1 != channel_name_2
 
 
+@pytest.mark.asyncio
+async def test_loop_instance_channel_layer_reference(channel_layer):
+    redis_pub_sub_loop_layer = channel_layer._get_layer()
+
+    assert redis_pub_sub_loop_layer.channel_layer == channel_layer
+
+
+def test_serialize(channel_layer):
+    """
+    Test default serialization method
+    """
+    message = {"a": True, "b": None, "c": {"d": []}}
+    serialized = channel_layer.serialize(message)
+    assert isinstance(serialized, bytes)
+    assert serialized == b"\x83\xa1a\xc3\xa1b\xc0\xa1c\x81\xa1d\x90"
+
+
+def test_deserialize(channel_layer):
+    """
+    Test default deserialization method
+    """
+    message = b"\x83\xa1a\xc3\xa1b\xc0\xa1c\x81\xa1d\x90"
+    deserialized = channel_layer.deserialize(message)
+
+    assert isinstance(deserialized, dict)
+    assert deserialized == {"a": True, "b": None, "c": {"d": []}}
+
+
 def test_multi_event_loop_garbage_collection(channel_layer):
     """
     Test loop closure layer flushing and garbage collection

tests/test_pubsub_sentinel.py

@@ -130,6 +130,34 @@ async def test_random_reset__channel_name(channel_layer):
     assert channel_name_1 != channel_name_2
 
 
+@pytest.mark.asyncio
+async def test_loop_instance_channel_layer_reference(channel_layer):
+    redis_pub_sub_loop_layer = channel_layer._get_layer()
+
+    assert redis_pub_sub_loop_layer.channel_layer == channel_layer
+
+
+def test_serialize(channel_layer):
+    """
+    Test default serialization method
+    """
+    message = {"a": True, "b": None, "c": {"d": []}}
+    serialized = channel_layer.serialize(message)
+    assert isinstance(serialized, bytes)
+    assert serialized == b"\x83\xa1a\xc3\xa1b\xc0\xa1c\x81\xa1d\x90"
+
+
+def test_deserialize(channel_layer):
+    """
+    Test default deserialization method
+    """
+    message = b"\x83\xa1a\xc3\xa1b\xc0\xa1c\x81\xa1d\x90"
+    deserialized = channel_layer.deserialize(message)
+
+    assert isinstance(deserialized, dict)
+    assert deserialized == {"a": True, "b": None, "c": {"d": []}}
+
+
 def test_multi_event_loop_garbage_collection(channel_layer):
     """
     Test loop closure layer flushing and garbage collection

tests/test_sentinel.py

@@ -644,3 +644,26 @@ def test_receive_buffer_respects_capacity():
     assert buff.qsize() == capacity
     messages = [buff.get_nowait() for _ in range(capacity)]
     assert list(range(9900, 10000)) == messages
+
+
+def test_serialize():
+    """
+    Test default serialization method
+    """
+    message = {"a": True, "b": None, "c": {"d": []}}
+    channel_layer = RedisChannelLayer()
+    serialized = channel_layer.serialize(message)
+    assert isinstance(serialized, bytes)
+    assert serialized[12:] == b"\x83\xa1a\xc3\xa1b\xc0\xa1c\x81\xa1d\x90"
+
+
+def test_deserialize():
+    """
+    Test default deserialization method
+    """
+    message = b"Q\x0c\xbb?Q\xbc\xe3|D\xfd9\x00\x83\xa1a\xc3\xa1b\xc0\xa1c\x81\xa1d\x90"
+    channel_layer = RedisChannelLayer()
+    deserialized = channel_layer.deserialize(message)
+
+    assert isinstance(deserialized, dict)
+    assert deserialized == {"a": True, "b": None, "c": {"d": []}}