[feature] Added a WebSocket endpoint for broadcasting all location updates #191

- Added a new Channels consumer at ``/ws/loci/location/all/`` to
  broadcast updates for all locations in a single feed.
- Implemented ``BaseCommonLocationBroadcast`` and ``CommonLocationBroadcast``.
- Updated signal receivers to publish to both per-location and
  common broadcast groups.
- Switched ``CHANNEL_LAYERS`` in the test project to use Redis.
- Updated local tooling and CI to include Redis service.

Closes #191

Author: dee077

.github/workflows/ci.yml

@@ -13,6 +13,12 @@ jobs:
     name: Python==${{ matrix.python-version }} | ${{ matrix.django-version }}
     runs-on: ubuntu-24.04
 
+    services:
+      redis:
+        image: redis
+        ports:
+          - 6379:6379
+
     strategy:
       fail-fast: false
       matrix:

README.rst

@@ -53,7 +53,7 @@ Reusable django-app for storing GIS and indoor coordinates of objects.
 Dependencies
 ------------
 
-- Python >= 3.8
+- Python >= 3.10
 - GeoDjango (`see GeoDjango Install Instructions
   <https://docs.djangoproject.com/en/dev/ref/contrib/gis/install/#requirements>`_)
 - One of the databases supported by GeoDjango
@@ -67,7 +67,7 @@ django-loci Python version
 0.3 - 0.4   >=3.6
 1.0         >=3.7
 1.1         >=3.8
-dev         >=3.8
+dev         >=3.10
 =========== ==============
 
 Install stable version from pypi
@@ -142,7 +142,10 @@ configuration can be:
     ASGI_APPLICATION = "django_loci.channels.asgi.channel_routing"
     CHANNEL_LAYERS = {
         "default": {
-            "BACKEND": "channels.layers.InMemoryChannelLayer",
+            "BACKEND": "channels_redis.core.RedisChannelLayer",
+            "CONFIG": {
+                "hosts": [("127.0.0.1", 6379)],
+            },
         },
     }
 
@@ -482,6 +485,17 @@ Extend the channel consumer of django-loci in this way:
     class LocationBroadcast(BaseLocationBroadcast):
         model = Location
 
+Extend the broadcast consumer for all locations:
+
+.. code-block:: python
+
+    from django_loci.channels.base import BaseCommonLocationBroadcast
+    from ..models import Location  # your own location model
+
+
+    class CommonLocationBroadcast(BaseCommonLocationBroadcast):
+        model = Location
+
 Extending AppConfig
 ~~~~~~~~~~~~~~~~~~~
 
@@ -524,6 +538,12 @@ Install test requirements:
 
     pip install -r requirements-test.txt
 
+Launch Redis:
+
+.. code-block:: shell
+
+    docker compose up -d redis
+
 Create database:
 
 .. code-block:: shell

django_loci/channels/asgi.py

@@ -4,8 +4,11 @@
 from django.core.asgi import get_asgi_application
 from django.urls import path
 
-from django_loci.channels.base import location_broadcast_path
-from django_loci.channels.consumers import LocationBroadcast
+from django_loci.channels.base import (
+    common_location_broadcast_path,
+    location_broadcast_path,
+)
+from django_loci.channels.consumers import CommonLocationBroadcast, LocationBroadcast
 
 channel_routing = ProtocolTypeRouter(
     {
@@ -17,7 +20,12 @@
                             location_broadcast_path,
                             LocationBroadcast.as_asgi(),
                             name="LocationChannel",
-                        )
+                        ),
+                        path(
+                            common_location_broadcast_path,
+                            CommonLocationBroadcast.as_asgi(),
+                            name="AllLocationChannel",
+                        ),
                     ]
                 )
             )

django_loci/channels/base.py

@@ -3,6 +3,7 @@
 from django.core.exceptions import ValidationError
 
 location_broadcast_path = "ws/loci/location/<uuid:pk>/"
+common_location_broadcast_path = "ws/loci/location/"
 
 
 def _get_object_or_none(model, **kwargs):
@@ -14,11 +15,15 @@ def _get_object_or_none(model, **kwargs):
 
 class BaseLocationBroadcast(JsonWebsocketConsumer):
     """
-    Notifies that the coordinates of a location have changed
-    to authorized users (superusers or organization operators)
+    Base WebSocket consumer for broadcasting location coordinate changes
+    to authorized users (superusers or organization operators).
     """
 
     def connect(self):
+        """
+        Handle WebSocket connection: authenticate user, validate location,
+        and join the location-specific broadcast group.
+        """
         self.pk = None
         try:
             user = self.scope["user"]
@@ -41,6 +46,10 @@ def connect(self):
             )
 
     def is_authorized(self, user, location):
+        """
+        Check if the user has permission to receive location broadcasts.
+        Requires authentication and change or view permissions on the location.
+        """
         perm = "{0}.change_location".format(self.model._meta.app_label)
         # allow users with view permission
         readperm = "{0}.view_location".format(self.model._meta.app_label)
@@ -49,15 +58,45 @@ def is_authorized(self, user, location):
         return authenticated and (user.is_superuser or (user.is_staff and is_permitted))
 
     def send_message(self, event):
+        """
+        Send JSON event data to the connected WebSocket client.
+        """
         self.send_json(event["message"])
 
     def disconnect(self, close_code):
         """
-        Perform things on connection close
+        Handle cleanup on WebSocket disconnection.
         """
         # The group_name is set only when the connection is accepted.
         # Remove the user from the group, if it exists.
         if hasattr(self, "group_name"):
             async_to_sync(self.channel_layer.group_discard)(
                 self.group_name, self.channel_name
             )
+
+
+class BaseCommonLocationBroadcast(BaseLocationBroadcast):
+
+    def connect(self):
+        """
+        Override connect to handle subscription to all locations
+        without requiring a specific location PK.
+        """
+        try:
+            user = self.scope["user"]
+        except KeyError:
+            self.close()
+        else:
+            if not self.is_authorized(user, None):
+                self.close()
+                return
+            self.accept()
+            self.join_groups(user)
+
+    def join_groups(self, user):
+        """
+        Subscribe to broadcast groups.
+        Subclasses can override to add user-specific groups (using the ``user`` argument).
+        """
+        self.group_name = "loci.mobile-location.common"
+        async_to_sync(self.channel_layer.group_add)(self.group_name, self.channel_name)

django_loci/channels/consumers.py

@@ -1,6 +1,10 @@
 from ..models import Location
-from .base import BaseLocationBroadcast
+from .base import BaseCommonLocationBroadcast, BaseLocationBroadcast
 
 
 class LocationBroadcast(BaseLocationBroadcast):
     model = Location
+
+
+class CommonLocationBroadcast(BaseCommonLocationBroadcast):
+    model = Location

django_loci/channels/receivers.py

@@ -7,15 +7,40 @@
 
 
 def update_mobile_location(sender, instance, **kwargs):
+    """
+    Sends WebSocket updates when a location record is updated.
+    - Sends a message to the location specific group.
+    - Sends a message to a common group for tracking all mobile location updates.
+    """
     if not kwargs.get("created") and instance.geometry:
-        group_name = "loci.mobile-location.{0}".format(str(instance.pk))
         channel_layer = channels.layers.get_channel_layer()
-        message = {
-            "geometry": json.loads(instance.geometry.geojson),
-            "address": instance.address,
-        }
+
+        # Send update to location specific group
+        async_to_sync(channel_layer.group_send)(
+            f"loci.mobile-location.{instance.pk}",
+            {
+                "type": "send_message",
+                "message": {
+                    "geometry": json.loads(instance.geometry.geojson),
+                    "address": instance.address,
+                },
+            },
+        )
+
+        # Send update to common mobile location group
         async_to_sync(channel_layer.group_send)(
-            group_name, {"type": "send_message", "message": message}
+            "loci.mobile-location.common",
+            {
+                "type": "send_message",
+                "message": {
+                    "id": str(instance.pk),
+                    "geometry": json.loads(instance.geometry.geojson),
+                    "address": instance.address,
+                    "name": instance.name,
+                    "type": instance.type,
+                    "is_mobile": instance.is_mobile,
+                },
+            },
         )
 
 

django_loci/tests/__init__.py

@@ -2,10 +2,15 @@
 Reusable test helpers
 """
 
+import importlib
 import os
 
+from channels.db import database_sync_to_async
+from channels.testing import WebsocketCommunicator
 from django.conf import settings
+from django.contrib.auth import login
 from django.core.files.uploadedfile import SimpleUploadedFile
+from django.http.request import HttpRequest
 
 
 class TestLociMixin(object):
@@ -137,3 +142,75 @@ def add_url(self):
     @property
     def change_url(self):
         return "{0}_change".format(self._get_url_prefix())
+
+
+class TestChannelsMixin(object):
+
+    async def _force_login(self, user, backend=None):
+        engine = importlib.import_module(settings.SESSION_ENGINE)
+        request = HttpRequest()
+        request.session = engine.SessionStore()
+        await database_sync_to_async(login)(request, user, backend)
+        await database_sync_to_async(request.session.save)()
+        return request.session
+
+    async def _get_location_request_dict(self, path, pk=None, user=None):
+        if not pk:
+            location = await database_sync_to_async(self._create_location)(
+                is_mobile=True
+            )
+            await database_sync_to_async(self._create_object_location)(
+                location=location
+            )
+            pk = location.pk
+        session = None
+        if user:
+            session = await self._force_login(user)
+        return {"pk": pk, "path": path, "session": session}
+
+    async def _get_specific_location_request_dict(self, pk=None, user=None):
+        result = await self._get_location_request_dict(
+            path="/ws/loci/location/{0}/", pk=pk, user=user
+        )
+        result["path"] = result["path"].format(result["pk"])
+        return result
+
+    async def _get_common_location_request_dict(self, pk=None, user=None):
+        return await self._get_location_request_dict(
+            path="/ws/loci/location/", pk=pk, user=user
+        )
+
+    def _get_location_communicator(
+        self, consumer, request_vars, user=None, include_pk=False
+    ):
+        communicator = WebsocketCommunicator(consumer.as_asgi(), request_vars["path"])
+        if user:
+            scope = {
+                "user": user,
+                "session": request_vars["session"],
+            }
+            if include_pk:
+                scope["url_route"] = {"kwargs": {"pk": request_vars["pk"]}}
+            communicator.scope.update(scope)
+        return communicator
+
+    def _get_specific_location_communicator(self, request_vars, user=None):
+        return self._get_location_communicator(
+            consumer=self.location_consumer,
+            request_vars=request_vars,
+            user=user,
+            include_pk=True,
+        )
+
+    def _get_common_location_communicator(self, request_vars, user=None):
+        return self._get_location_communicator(
+            consumer=self.common_location_consumer,
+            request_vars=request_vars,
+            user=user,
+            include_pk=False,
+        )
+
+    async def _save_location(self, pk):
+        loc = await self.location_model.objects.aget(pk=pk)
+        loc.geometry = "POINT (12.513124 41.897903)"
+        await loc.asave()