feat: websockets

backoffice/consumers.py

@@ -0,0 +1,27 @@
+import json
+
+from channels.generic.websocket import AsyncWebsocketConsumer
+from django.utils.html import escape
+
+
+class EchoConsumer(AsyncWebsocketConsumer):
+    """Consumer mínimo para verificar el cableado de Channels + htmx (ver
+    ws_demo.html). No representa ninguna funcionalidad real todavía."""
+
+    async def connect(self):
+        # Mismo criterio de acceso que BackofficeAccessMixin.
+        if not self.scope['user'].is_staff:
+            await self.close()
+            return
+
+        await self.accept()
+
+    async def receive(self, text_data):
+        try:
+            payload = json.loads(text_data)
+        except (TypeError, ValueError):
+            payload = {}
+
+        message = escape(payload.get('message', ''))
+
+        await self.send(text_data=f'<div id="ws-demo-log" hx-swap-oob="beforeend"><p>Eco: {message}</p></div>')