Add observer for signals

Author: hishnash

channels_api/observer.py

@@ -0,0 +1,67 @@
+from functools import partial
+from typing import Dict, Any
+
+from asgiref.sync import async_to_sync
+from channels.consumer import AsyncConsumer
+from channels.layers import get_channel_layer
+from django.dispatch import Signal
+
+
+class Observer:
+    def __init__(self, func, signal: Signal=None, kwargs=None):
+        if kwargs is None:
+            kwargs = {}
+        self.func = func
+        self.signal = signal
+        self.signal_kwargs = kwargs
+        self._serializer = None
+        self.signal.connect(
+            self.handle, **self.signal_kwargs
+        )
+
+    async def __call__(self, *args, **kwargs):
+        return await self.func(*args, **kwargs)
+
+    def serialize(self, signal, *args, **kwargs) -> Dict[str, Any]:
+        message = {}
+        if self._serializer:
+            message = self._serializer(self, signal, *args, **kwargs)
+        message['type'] = self.func.__name__.replace('_', '.')
+
+        return message
+
+    def serializer(self, func):
+        self._serializer = func
+
+    def handle(self, signal, *args, **kwargs):
+        message = self.serialize(signal, *args, **kwargs)
+        channel_layer = get_channel_layer()
+        group_name = self.channel_name(signal, *args, **kwargs)
+        async_to_sync(channel_layer.group_send)(group_name, message)
+
+    def channel_name(self, *args, **kwargs):
+        return '{}-signal-{}'.format(
+            self.func.__name__.replace('_', '.'),
+            '.'.join(
+                arg.lower().replace('_', '.') for arg in
+                self.signal.providing_args
+            )
+        )
+
+    def __get__(self, parent, objtype):
+
+        if parent is None:
+            return self
+
+        return partial(self.__call__, parent)
+
+    async def subscribe(self, consumer: AsyncConsumer, *args, **kwargs):
+        await consumer.channel_layer.group_add(
+            self.channel_name(*args, **kwargs),
+            consumer.channel_name
+        )
+        print('subscribed', consumer, self.channel_name(*args, **kwargs), consumer.channel_name)
+
+
+def observer(signal, **kwargs):
+    return partial(Observer, signal=signal, kwargs=kwargs)

requirements.txt

@@ -1,5 +1,5 @@
 django>=1.8
 djangorestframework>=3
-channels>=2.0.2
+git+https://github.com/django/channels.git
 pytest-django
 pytest-asyncio

setup.py

@@ -25,6 +25,9 @@
             'coverage~=4.4',
         ],
     },
+    dependency_links=[
+        'git+https://github.com/django/channels.git'
+    ],
     classifiers=[
         'Programming Language :: Python :: 3',
         'Programming Language :: Python :: 3.4',

tests/test_consumer.py

@@ -25,7 +25,9 @@ def test_sync_action(self, pk=None):
 
     # Test a normal connection
     communicator = WebsocketCommunicator(AConsumer, "/testws/")
+
     connected, _ = await communicator.connect()
+
     assert connected
 
     await communicator.send_json_to(
@@ -63,3 +65,5 @@ def test_sync_action(self, pk=None):
         'response_status': 200,
         'request_id': 10
     }
+
+    await communicator.disconnect()

tests/test_observer.py

@@ -0,0 +1,61 @@
+import pytest
+from channels import DEFAULT_CHANNEL_LAYER
+from channels.db import database_sync_to_async
+from channels.generic.websocket import AsyncJsonWebsocketConsumer
+from channels.layers import channel_layers
+from channels.testing import WebsocketCommunicator
+from django.contrib.auth import user_logged_in, get_user_model
+
+from channels_api.observer import observer
+
+
+@pytest.mark.django_db(transaction=True)
+@pytest.mark.asyncio
+async def test_observer_wrapper(settings):
+    settings.CHANNEL_LAYERS={
+        "default": {
+            "BACKEND": "channels.layers.InMemoryChannelLayer",
+            "TEST_CONFIG": {
+                "expiry": 100500,
+            },
+        },
+    }
+
+    layer = channel_layers.make_test_backend(DEFAULT_CHANNEL_LAYER)
+
+    class TestConsumer(AsyncJsonWebsocketConsumer):
+
+        async def dispatch(self, message):
+            """
+            Works out what to do with a message.
+            """
+            await super().dispatch(message)
+
+        async def accept(self):
+            await TestConsumer.handle_user_logged_in.subscribe(self)
+            await super().accept()
+
+        @observer(user_logged_in)
+        async def handle_user_logged_in(self, *args, **kwargs):
+            await self.send_json({'message': kwargs,})
+
+    communicator = WebsocketCommunicator(TestConsumer, "/testws/")
+
+    connected, _ = await communicator.connect()
+
+    assert connected
+
+    user = await database_sync_to_async(get_user_model().objects.create)(
+        username='test',
+        email='[email protected]'
+    )
+
+    await database_sync_to_async(user_logged_in.send)(
+        sender=user.__class__,
+        request=None,
+        user=user
+    )
+
+    response = await communicator.receive_json_from()
+
+    assert {'message': {}} == response