Why not rewrite binding into multiplexers on a Monday night?

channels/binding/base.py

@@ -120,11 +120,10 @@ class Binding(object):
     # Inbound binding
 
     @classmethod
-    def trigger_inbound(cls, message):
+    def trigger_inbound(cls, message, **kwargs):
         """
         Triggers the binding to see if it will do something.
-        We separate out message serialization to a consumer, so this gets
+        Also acts as a consumer.
-        native arguments.
         """
         # Late import as it touches models
         from django.contrib.auth.models import AnonymousUser
@@ -136,6 +135,8 @@ class Binding(object):
         # Run incoming action
         self.run_action(self.action, self.pk, self.data)
 
+    consumer = trigger_inbound
+
     def deserialize(self, message):
         """
         Returns action, pk, data decoded from the message. pk should be None

channels/binding/websockets.py

@@ -3,7 +3,7 @@ import json
 from django.core import serializers
 
 from .base import Binding
-from ..generic.websockets import JsonWebsocketConsumer
+from ..generic.websockets import JsonWebsocketConsumer, WebsocketDemultiplexer
 
 
 class WebsocketBinding(Binding):
@@ -26,19 +26,24 @@ class WebsocketBinding(Binding):
 
     model = None
 
+    # Optional stream multiplexing
+
+    stream = None
+
     # Outbound
 
     def serialize(self, instance, action):
+        payload = {
+            "action": action,
+            "pk": instance.pk,
+            "data": self.serialize_data(instance),
+        }
+        # Encode for the stream
+        assert self.stream is not None
+        payload = WebsocketDemultiplexer.encode(self.stream, payload)
+        # Return WS format message
         return {
-            "text": json.dumps({
+            "text": json.dumps(payload),
-                "model": "%s.%s" % (
-                    instance._meta.app_label.lower(),
-                    instance._meta.object_name.lower(),
-                ),
-                "action": action,
-                "pk": instance.pk,
-                "data": self.serialize_data(instance),
-            }),
         }
 
     def serialize_data(self, instance):
@@ -51,10 +56,13 @@ class WebsocketBinding(Binding):
     # Inbound
 
     def deserialize(self, message):
-        content = json.loads(message['text'])
+        """
-        action = content['action']
+        You must hook this up behind a Deserializer, so we expect the JSON
-        pk = content.get('pk', None)
+        already dealt with.
-        data = content.get('data', None)
+        """
+        action = message['action']
+        pk = message.get('pk', None)
+        data = message.get('data', None)
         return action, pk, data
 
     def _hydrate(self, pk, data):
@@ -81,29 +89,3 @@ class WebsocketBinding(Binding):
         for name in data.keys():
             setattr(instance, name, getattr(hydrated.object, name))
         instance.save()
-
-
-class WebsocketBindingDemultiplexer(JsonWebsocketConsumer):
-    """
-    Allows you to combine multiple Bindings as one websocket consumer.
-    Subclass and provide a custom list of Bindings.
-    """
-
-    http_user = True
-    warn_if_no_match = True
-    bindings = None
-
-    def receive(self, content):
-        # Sanity check
-        if self.bindings is None:
-            raise ValueError("Demultiplexer has no bindings!")
-        # Find the matching binding
-        model_label = content['model']
-        triggered = False
-        for binding in self.bindings:
-            if binding.model_label == model_label:
-                binding.trigger_inbound(self.message)
-                triggered = True
-        # At least one of them should have fired.
-        if not triggered and self.warn_if_no_match:
-            raise ValueError("No binding found for model %s" % model_label)

channels/generic/websockets.py

@@ -1,6 +1,6 @@
 import json
 
-from ..channel import Group
+from ..channel import Group, Channel
 from ..auth import channel_session_user_from_http
 from ..sessions import enforce_ordering
 from .base import BaseConsumer
@@ -155,3 +155,57 @@ class JsonWebsocketConsumer(WebsocketConsumer):
 
     def group_send(self, name, content):
         super(JsonWebsocketConsumer, self).group_send(name, json.dumps(content))
+
+
+class WebsocketDemultiplexer(JsonWebsocketConsumer):
+    """
+    JSON-understanding WebSocket consumer subclass that handles demultiplexing
+    streams using a "stream" key in a top-level dict and the actual payload
+    in a sub-dict called "payload". This lets you run multiple streams over
+    a single WebSocket connection in a standardised way.
+
+    Incoming messages on streams are mapped into a custom channel so you can
+    just tie in consumers the normal way. The reply_channels are kept so
+    sessions/auth continue to work. Payloads must be a dict at the top level,
+    so they fulfill the Channels message spec.
+
+    Set a mapping from streams to channels in the "mapping" key. We make you
+    whitelist channels like this to allow different namespaces and for security
+    reasons (imagine if someone could inject straight into websocket.receive).
+    """
+
+    mapping = {}
+
+    def receive(self, content, **kwargs):
+        # Check the frame looks good
+        if isinstance(content, dict) and "stream" in content and "payload" in content:
+            # Match it to a channel
+            stream = content['stream']
+            if stream in self.mapping:
+                # Extract payload and add in reply_channel
+                payload = content['payload']
+                if not isinstance(payload, dict):
+                    raise ValueError("Multiplexed frame payload is not a dict")
+                payload['reply_channel'] = self.message['reply_channel']
+                # Send it onto the new channel
+                Channel(self.mapping[stream]).send(payload)
+            else:
+                raise ValueError("Invalid multiplexed frame received (stream not mapped)")
+        else:
+            raise ValueError("Invalid multiplexed frame received (no channel/payload key)")
+
+    def send(self, stream, payload):
+        super(WebsocketDemultiplexer, self).send(self.encode(stream, payload))
+
+    def group_send(self, name, stream, payload):
+        super(WebsocketDemultiplexer, self).group_send(name, self.encode(stream, payload))
+
+    @classmethod
+    def encode(cls, stream, payload):
+        """
+        Encodes stream + payload for outbound sending.
+        """
+        return {
+            "stream": stream,
+            "payload": payload,
+        }

docs/binding.rst

@@ -1,6 +1,11 @@
 Data Binding
 ============
 
+.. warning::
+
+    The Data Binding part is new and might change slightly in the
+    upcoming weeks, and so don't consider this API totally stable yet.
+
 The Channels data binding framework automates the process of tying Django
 models into frontend views, such as javascript-powered website UIs. It provides
 a quick and flexible way to generate messages on Groups for model changes