Allow the subscription broadcast job queue to be configured (#1301)

CHANGELOG.md

@@ -29,6 +29,7 @@ You can find and compare releases at the [GitHub release page](https://github.co
 - Report errors that are not client-safe through Laravel's `ExceptionHandler` https://github.com/nuwave/lighthouse/pull/1303
 - Log in subscribers when broadcasting a subscription update, so that calls to `auth()->user()` return
   the authenticated user instead of `null` https://github.com/nuwave/lighthouse/pull/1306
+- Replace the subscription broadcast queued event handler with a queued job to allow the queue name to be specified https://github.com/nuwave/lighthouse/pull/1301
 
 ## 4.11.0
 

UPGRADE.md

@@ -116,3 +116,13 @@ The method will have to change like this:
 -public function purchasedItemsCount($root, array $args)
 +public function purchasedItemsCount(int $year, ?bool $includeReturns)
 ```
+
+### `Nuwave\Lighthouse\Subscriptions\Events\BroadcastSubscriptionEvent` is no longer fired
+
+The event is no longer fired, and the event class was removed. Lighthouse now uses a queued job instead.
+
+If you manually fired the event, replace it by queuing a `Nuwave\Lighthouse\Subscriptions\BroadcastSubscriptionJob`
+or a call to `Nuwave\Lighthouse\Subscriptions\Contracts\BroadcastsSubscriptions::queueBroadcast()`.
+
+In case you depend on an event being fired whenever a subscription is queued, you can bind your
+own implementation of `Nuwave\Lighthouse\Subscriptions\Contracts\BroadcastsSubscriptions`.

composer.json

@@ -27,6 +27,7 @@
         "ext-json": "*",
         "nesbot/carbon": "^1.26.0 || ^2.0",
         "illuminate/auth": "5.5.* || 5.6.* || 5.7.* || 5.8.* || ^6.0 || ^7.0",
+        "illuminate/bus": "5.5.* || 5.6.* || 5.7.* || 5.8.* || ^6.0 || ^7.0",
         "illuminate/contracts": "5.5.* || 5.6.* || 5.7.* || 5.8.* || ^6.0 || ^7.0",
         "illuminate/http": "5.5.* || 5.6.* || 5.7.* || 5.8.* || ^6.0 || ^7.0",
         "illuminate/pagination": "5.5.* || 5.6.* || 5.7.* || 5.8.* || ^6.0 || ^7.0",

src/Execution/Utils/Subscription.php

@@ -14,7 +14,6 @@ class Subscription
     /**
      * Broadcast subscription to client(s).
      *
-     *
      * @throws \InvalidArgumentException
      */
     public static function broadcast(string $subscriptionField, $root, ?bool $shouldQueue = null): void
@@ -35,25 +34,22 @@ public static function broadcast(string $subscriptionField, $root, ?bool $should
         /** @var \Nuwave\Lighthouse\Subscriptions\Contracts\BroadcastsSubscriptions $broadcaster */
         $broadcaster = app(BroadcastsSubscriptions::class);
 
-        $shouldQueue = $shouldQueue === null
-            ? config('lighthouse.subscriptions.queue_broadcasts', false)
-            : $shouldQueue;
+        // Default to the configuration setting if not specified
+        if ($shouldQueue === null) {
+            $shouldQueue = config('lighthouse.subscriptions.queue_broadcasts', false);
+        }
 
-        $method = $shouldQueue
-            ? 'queueBroadcast'
-            : 'broadcast';
+        $subscription = $registry->subscription($subscriptionField);
 
         try {
-            call_user_func(
-                [$broadcaster, $method],
-                $registry->subscription($subscriptionField),
-                $subscriptionField,
-                $root
-            );
+            if ($shouldQueue) {
+                $broadcaster->queueBroadcast($subscription, $subscriptionField, $root);
+            } else {
+                $broadcaster->broadcast($subscription, $subscriptionField, $root);
+            }
         } catch (Throwable $e) {
             /** @var \Nuwave\Lighthouse\Subscriptions\Contracts\SubscriptionExceptionHandler $exceptionHandler */
             $exceptionHandler = app(SubscriptionExceptionHandler::class);
-
             $exceptionHandler->handleBroadcastError($e);
         }
     }

src/Subscriptions/BroadcastSubscriptionJob.php

@@ -0,0 +1,45 @@
+<?php
+
+namespace Nuwave\Lighthouse\Subscriptions;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Queue\SerializesModels;
+use Nuwave\Lighthouse\Schema\Types\GraphQLSubscription;
+use Nuwave\Lighthouse\Subscriptions\Contracts\BroadcastsSubscriptions;
+
+class BroadcastSubscriptionJob implements ShouldQueue
+{
+    use Queueable, SerializesModels;
+
+    /**
+     * The subscription field that was requested.
+     *
+     * @var \Nuwave\Lighthouse\Schema\Types\GraphQLSubscription
+     */
+    public $subscription;
+
+    /**
+     * The name of the field.
+     *
+     * @var string
+     */
+    public $fieldName;
+
+    /**
+     * The root element to be passed when resolving the subscription.
+     */
+    public $root;
+
+    public function __construct(GraphQLSubscription $subscription, string $fieldName, $root)
+    {
+        $this->subscription = $subscription;
+        $this->fieldName = $fieldName;
+        $this->root = $root;
+    }
+
+    public function handle(BroadcastsSubscriptions $broadcaster): void
+    {
+        $broadcaster->broadcast($this->subscription, $this->fieldName, $this->root);
+    }
+}

src/Subscriptions/Events/BroadcastSubscriptionEvent.php

@@ -3,8 +3,11 @@
 namespace Nuwave\Lighthouse\Subscriptions\Events;
 
 use Illuminate\Queue\SerializesModels;
-use Nuwave\Lighthouse\Schema\Types\GraphQLSubscription as Subscription;
+use Nuwave\Lighthouse\Schema\Types\GraphQLSubscription;
 
+/**
+ * @deprecated will be removed in v5 and replaced with a Job
+ */
 class BroadcastSubscriptionEvent
 {
     use SerializesModels;
@@ -21,7 +24,7 @@ class BroadcastSubscriptionEvent
 
     public $root;
 
-    public function __construct(Subscription $subscription, string $fieldName, $root)
+    public function __construct(GraphQLSubscription $subscription, string $fieldName, $root)
     {
         $this->subscription = $subscription;
         $this->fieldName = $fieldName;

src/Subscriptions/Events/BroadcastSubscriptionListener.php

@@ -2,19 +2,23 @@
 
 namespace Nuwave\Lighthouse\Subscriptions\Events;
 
-use Illuminate\Contracts\Queue\ShouldQueue;
-use Nuwave\Lighthouse\Subscriptions\Contracts\BroadcastsSubscriptions;
+use Illuminate\Contracts\Bus\Dispatcher as BusDispatcher;
+use Nuwave\Lighthouse\Subscriptions\BroadcastSubscriptionJob;
 
-class BroadcastSubscriptionListener implements ShouldQueue
+/**
+ * @deprecated This class is just here to preserve backwards compatiblity with v4.
+ * TODO remove the event and handle subscriptions as commands only in v5
+ */
+class BroadcastSubscriptionListener
 {
     /**
-     * @var \Nuwave\Lighthouse\Subscriptions\Contracts\BroadcastsSubscriptions
+     * @var \Illuminate\Contracts\Events\Dispatcher
      */
-    protected $broadcaster;
+    protected $busDispatcher;
 
-    public function __construct(BroadcastsSubscriptions $broadcaster)
+    public function __construct(BusDispatcher $busDispatcher)
     {
-        $this->broadcaster = $broadcaster;
+        $this->busDispatcher = $busDispatcher;
     }
 
     /**
@@ -24,10 +28,12 @@ public function __construct(BroadcastsSubscriptions $broadcaster)
      */
     public function handle(BroadcastSubscriptionEvent $event): void
     {
-        $this->broadcaster->broadcast(
-            $event->subscription,
-            $event->fieldName,
-            $event->root
+        $this->busDispatcher->dispatch(
+            (new BroadcastSubscriptionJob(
+                $event->subscription,
+                $event->fieldName,
+                $event->root
+            ))->onQueue(config('lighthouse.subscriptions.broadcasts_queue_name', null))
         );
     }
 }

src/Subscriptions/SubscriptionBroadcaster.php

@@ -66,6 +66,7 @@ public function __construct(
      */
     public function queueBroadcast(GraphQLSubscription $subscription, string $fieldName, $root): void
     {
+        // TODO replace with a job dispatch
         $this->eventsDispatcher->dispatch(
             new BroadcastSubscriptionEvent($subscription, $fieldName, $root)
         );

src/lighthouse.php

@@ -247,6 +247,11 @@
          */
         'queue_broadcasts' => env('LIGHTHOUSE_QUEUE_BROADCASTS', true),
 
+        /*
+         * Determines the queue to use for broadcasting queue jobs.
+         */
+        'broadcasts_queue_name' => env('LIGHTHOUSE_BROADCASTS_QUEUE_NAME', null),
+
         /*
          * Default subscription storage.
          *
@@ -255,7 +260,7 @@
         'storage' => env('LIGHTHOUSE_SUBSCRIPTION_STORAGE', 'redis'),
 
         /*
-         * Default subscription storage time to live.
+         * Default subscription storage time to live in seconds.
          *
          * Indicates how long a subscription can be active before it's automatically removed from storage.
          * Setting this to `null` means the subscriptions are stored forever. This may cause