Introduce initial payload.

Author: Andrey Zelenchuk

channels_graphql_ws/graphql_ws_consumer.py

@@ -679,6 +679,7 @@ async def _register_subscription(
         groups,
         publish_callback,
         unsubscribed_callback,
+        initial_payload,
         notification_queue_limit=None,
     ):
         """Register a new subscription when client subscribes.
@@ -697,6 +698,7 @@ async def _register_subscription(
                 subscription groups current subscription belongs to.
             unsubscribed_callback: Called to notify when a client
                 unsubscribes.
+            initial_payload: Initial payload.
             notification_queue_limit: LImit for the subscribtion
                 notification queue. Default is used if not set.
 
@@ -720,13 +722,25 @@ async def _register_subscription(
             queue_size = self.subscription_notification_queue_limit
         notification_queue = asyncio.Queue(maxsize=queue_size)
 
+        # Enqueue the initial payload.
+        if initial_payload is not self.SKIP:
+            notification_queue.put_nowait(Serializer.serialize(initial_payload))
+
         # Start an endless task which listens the `notification_queue`
         # and invokes subscription "resolver" on new notifications.
         async def notifier():
             """Watch the notification queue and notify clients."""
 
             # Assert we run in a proper thread.
             self._assert_thread()
+
+            # Dirty hack to partially workaround the race between:
+            #  1) call to `result.subscribe` in `_on_gql_start`; and
+            #  2) call to `trigger.on_next` below in this function.
+            # The first call must be earlier. Otherwise, first one or more notifications
+            # may be lost.
+            await asyncio.sleep(1)
+
             while True:
                 serialized_payload = await notification_queue.get()
 

channels_graphql_ws/subscription.py

@@ -148,6 +148,11 @@ class Subscription(graphene.ObjectType):
     # Return this from the `publish` to suppress the notification.
     SKIP = GraphqlWsConsumer.SKIP
 
+    # Initial payload. Set it to send an "initial" response with this
+    # payload to a client as soon as it is subscribed (before any call
+    # to `Subscription.broadcast`).
+    initial_payload = GraphqlWsConsumer.SKIP
+
     # Subscription notifications queue limit. Set this to control the
     # amount of notifications server keeps in queue when notifications
     # come faster than server processing them. Set this limit to 1 drops
@@ -435,6 +440,7 @@ def unsubscribed_callback():
             groups,
             publish_callback,
             unsubscribed_callback,
+            cls.initial_payload,
             cls.notification_queue_limit,
         )