Acknowledge PubSub event message

Author: richan-fongdasen

composer.json

@@ -52,7 +52,10 @@
     "laravel": {
       "providers": [
         "RichanFongdasen\\GCRWorker\\ServiceProvider"
-      ]
+      ],
+      "aliases": {
+        "GcrQueue": "RichanFongdasen\\GCRWorker\\Facade\\GcrQueue"
+      }
     }
   },
   "scripts": {

src/Concerns/CreatesPubSubQueue.php

@@ -0,0 +1,82 @@
+<?php
+
+namespace RichanFongdasen\GCRWorker\Concerns;
+
+use ErrorException;
+use Kainxspirits\PubSubQueue\Connectors\PubSubConnector;
+use Kainxspirits\PubSubQueue\PubSubQueue;
+
+trait CreatesPubSubQueue
+{
+    /**
+     * Pub/Sub connector instance.
+     *
+     * @var PubSubConnector
+     */
+    protected $connector;
+
+    /**
+     * The base PubSubQueue instance.
+     *
+     * @var PubSubQueue
+     */
+    protected $pubSub;
+
+    /**
+     * Create the PubSubQueue instance.
+     *
+     * @throws ErrorException
+     *
+     * @return PubSubQueue
+     */
+    private function createQueue(): PubSubQueue
+    {
+        $queue = $this->connector->connect(config('queue.connections.pubsub', ['project_id' => 'default-project-id']));
+
+        if (!($queue instanceof PubSubQueue)) {
+            throw new ErrorException('Failed to retrieve PubSubQueue instance.');
+        }
+
+        return $queue;
+    }
+
+    /**
+     * Retrieve the PubSubConnector instance.
+     *
+     * @return PubSubConnector
+     */
+    public function getConnector(): PubSubConnector
+    {
+        return $this->connector;
+    }
+
+    /**
+     * Retrieve the PubSubQueue instance.
+     *
+     * @return PubSubQueue
+     */
+    public function getPubSubQueue(): PubSubQueue
+    {
+        return $this->pubSub;
+    }
+
+    /**
+     * Initialize the PubSubQueue instance.
+     *
+     * @throws ErrorException
+     */
+    protected function initializeQueue(): void
+    {
+        $this->pubSub = $this->createQueue();
+    }
+
+    /**
+     * Set the PubSubConnector instance.
+     *
+     * @param PubSubConnector $connector
+     */
+    public function setConnector(PubSubConnector $connector): void
+    {
+        $this->connector = $connector;
+    }
+}

src/Facade/GcrQueue.php

@@ -0,0 +1,33 @@
+<?php
+
+namespace RichanFongdasen\GCRWorker\Facade;
+
+use Illuminate\Support\Facades\Facade;
+use RichanFongdasen\GCRWorker\GcrPubSubQueue;
+use RichanFongdasen\GCRWorker\GcrPubSubQueueFake;
+
+class GcrQueue extends Facade
+{
+    /**
+     * Replace the bound instance with a fake.
+     *
+     * @return GcrPubSubQueueFake
+     */
+    public static function fake(): GcrPubSubQueueFake
+    {
+        $fake = app(GcrPubSubQueueFake::class);
+        static::swap($fake);
+
+        return $fake;
+    }
+
+    /**
+     * Get the registered name of the component.
+     *
+     * @return string
+     */
+    protected static function getFacadeAccessor()
+    {
+        return GcrPubSubQueue::class;
+    }
+}

src/GcrPubSubQueue.php

@@ -0,0 +1,34 @@
+<?php
+
+namespace RichanFongdasen\GCRWorker;
+
+use Google\Cloud\PubSub\Message;
+use Kainxspirits\PubSubQueue\Connectors\PubSubConnector;
+use RichanFongdasen\GCRWorker\Concerns\CreatesPubSubQueue;
+
+class GcrPubSubQueue
+{
+    use CreatesPubSubQueue;
+
+    /**
+     * GcrPubSubQueue constructor.
+     *
+     * @param PubSubConnector $connector
+     * @throws \ErrorException
+     */
+    public function __construct(PubSubConnector $connector)
+    {
+        $this->setConnector($connector);
+        $this->initializeQueue();
+    }
+
+    /**
+     * Acknowledge the given message.
+     *
+     * @param Message $message
+     */
+    public function acknowledge(Message $message): void
+    {
+        $this->pubSub->acknowledge($message);
+    }
+}

src/GcrPubSubQueueFake.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace RichanFongdasen\GCRWorker;
+
+use Google\Cloud\PubSub\Message;
+use Illuminate\Support\Collection;
+use Kainxspirits\PubSubQueue\Connectors\PubSubConnector;
+use PHPUnit\Framework\Assert as PHPUnit;
+use RichanFongdasen\GCRWorker\Concerns\CreatesPubSubQueue;
+
+class GcrPubSubQueueFake
+{
+    use CreatesPubSubQueue;
+
+    /**
+     * The collection of acknowledged messages.
+     *
+     * @var Collection
+     */
+    protected $acknowledgedMessages;
+
+    /**
+     * GcrPubSubQueue constructor.
+     *
+     * @param PubSubConnector $connector
+     * @throws \ErrorException
+     */
+    public function __construct(PubSubConnector $connector)
+    {
+        $this->acknowledgedMessages = collect();
+
+        $this->setConnector($connector);
+        $this->initializeQueue();
+    }
+
+    /**
+     * Acknowledge the given message.
+     *
+     * @param Message $message
+     */
+    public function acknowledge(Message $message): void
+    {
+        $this->acknowledgedMessages->put($message->id(), $message);
+    }
+
+    /**
+     * Assert if the acknowledged messages count equals to the given value.
+     *
+     * @param int $count
+     */
+    public function assertAcknowledgedMessagesCount(int $count): void
+    {
+        PHPUnit::assertEquals($count, $this->acknowledgedMessages->count());
+    }
+
+    /**
+     * Assert if the given message id has been acknowledged.
+     *
+     * @param string $messageId
+     */
+    public function assertMessageHasAcknowledged(string $messageId): void
+    {
+        PHPUnit::assertTrue($this->acknowledgedMessages->has($messageId));
+    }
+}

src/PubSubEvent.php

@@ -2,49 +2,28 @@
 
 namespace RichanFongdasen\GCRWorker;
 
-use ErrorException;
 use Google\Cloud\PubSub\Message;
 use Illuminate\Container\Container;
-use Kainxspirits\PubSubQueue\Connectors\PubSubConnector;
 use Kainxspirits\PubSubQueue\Jobs\PubSubJob;
-use Kainxspirits\PubSubQueue\PubSubQueue;
+use RichanFongdasen\GCRWorker\Facade\GcrQueue;
 
 class PubSubEvent
 {
-    /**
-     * Pub/Sub connector instance.
-     *
-     * @var PubSubConnector
-     */
-    protected $connector;
-
     /**
      * Laravel IOC Container instance.
      *
      * @var Container
      */
     protected $container;
 
-    /**
-     * Pub/Sub queue instance.
-     *
-     * @var PubSubQueue
-     */
-    protected $queue;
-
     /**
      * PubSubEvent constructor.
      *
-     * @param Container       $container
-     * @param PubSubConnector $connector
-     *
-     * @throws ErrorException
+     * @param Container $container
      */
-    public function __construct(Container $container, PubSubConnector $connector)
+    public function __construct(Container $container)
     {
         $this->container = $container;
-        $this->connector = $connector;
-        $this->queue = $this->getQueue();
     }
 
     /**
@@ -58,31 +37,13 @@ protected function createJob(Message $message): PubSubJob
     {
         return new PubSubJob(
             $this->container,
-            $this->queue,
+            GcrQueue::getPubSubQueue(),
             $message,
             config('queue.connections.pubsub.driver'),
             config('queue.connections.pubsub.queue')
         );
     }
 
-    /**
-     * Get the PubSubQueue instance.
-     *
-     * @throws ErrorException
-     *
-     * @return PubSubQueue
-     */
-    protected function getQueue(): PubSubQueue
-    {
-        $queue = $this->connector->connect(config('queue.connections.pubsub'));
-
-        if (!($queue instanceof PubSubQueue)) {
-            throw new ErrorException('Failed to retrieve PubSubQueue instance.');
-        }
-
-        return $queue;
-    }
-
     /**
      * Handle the given Pub/Sub event message.
      *
@@ -92,6 +53,8 @@ public function handle(Message $message): void
     {
         set_time_limit(config('gcr-worker.max_execution_time'));
 
+        GcrQueue::acknowledge($message);
+
         $this->createJob($message)->fire();
     }
 }

tests/Feature/PubSubEventHandlingTest.php

@@ -3,6 +3,7 @@
 namespace RichanFongdasen\GCRWorker\Tests\Feature;
 
 use Illuminate\Support\Facades\Cache;
+use RichanFongdasen\GCRWorker\Facade\GcrQueue;
 use RichanFongdasen\GCRWorker\Tests\TestCase;
 
 class PubSubEventHandlingTest extends TestCase
@@ -19,6 +20,8 @@ public function it_will_return_403_on_unauthorized_pubsub_request()
     /** @test */
     public function it_can_handle_pubsub_invocation_as_expected()
     {
+        GcrQueue::fake();
+
         config(['gcr-worker.allow_event_invocation' => true]);
 
         $data = json_decode(file_get_contents(dirname(__DIR__, 2).'/dummies/message.json'), true);
@@ -29,6 +32,9 @@ public function it_can_handle_pubsub_invocation_as_expected()
             ->assertStatus(200)
             ->assertJsonFragment(['info' => 'The Pub/Sub queued job has completed.']);
 
+        GcrQueue::assertAcknowledgedMessagesCount(1);
+        GcrQueue::assertMessageHasAcknowledged('1777817206939726');
+
         self::assertEquals('completed', Cache::get('dummy-job-status'));
     }
 }

tests/TestCase.php

@@ -3,6 +3,7 @@
 namespace RichanFongdasen\GCRWorker\Tests;
 
 use Orchestra\Testbench\TestCase as BaseTest;
+use RichanFongdasen\GCRWorker\Facade\GcrQueue;
 
 abstract class TestCase extends BaseTest
 {
@@ -46,6 +47,7 @@ protected function getPackageAliases($app): array
 
         return [
             'Route' => \Illuminate\Support\Facades\Route::class,
+            'GcrQueue' => GcrQueue::class,
         ];
     }