Introduce initial payload.

Author: Andrey Zelenchuk

channels_graphql_ws/graphql_ws_consumer.py

@@ -665,7 +665,12 @@ def register_middleware(next_middleware, root, info, *args, **kwds):
                 await self._send_gql_complete(operation_id)
 
     async def _register_subscription(
-        self, operation_id, groups, publish_callback, unsubscribed_callback
+        self,
+        operation_id,
+        groups,
+        publish_callback,
+        unsubscribed_callback,
+        initial_payload,
     ):
         """Register a new subscription when client subscribes.
 
@@ -703,13 +708,25 @@ async def _register_subscription(
             maxsize=self.subscription_notification_queue_limit
         )
 
+        # 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

@@ -357,6 +357,7 @@ def __init_subclass_with_meta__(
         _meta.subscribe = get_function(subscribe)
         _meta.publish = get_function(publish)
         _meta.unsubscribed = get_function(unsubscribed)
+        _meta.initial_payload = options.get("initial_payload", cls.SKIP)
 
         super().__init_subclass_with_meta__(_meta=_meta, **options)
 
@@ -423,7 +424,9 @@ def unsubscribed_callback():
             # `subscribe`.
             return result
 
-        return register_subscription(groups, publish_callback, unsubscribed_callback)
+        return register_subscription(
+            groups, publish_callback, unsubscribed_callback, cls._meta.initial_payload
+        )
 
     @classmethod
     def _group_name(cls, group=None):