Pull fresh PubSub message from subscription to get a message with ackId

Author: richan-fongdasen

src/GcrPubSubQueue.php

@@ -2,6 +2,7 @@
 
 namespace RichanFongdasen\GCRWorker;
 
+use ErrorException;
 use Google\Cloud\PubSub\Message;
 use Kainxspirits\PubSubQueue\Connectors\PubSubConnector;
 use RichanFongdasen\GCRWorker\Concerns\CreatesPubSubQueue;
@@ -32,4 +33,30 @@ public function acknowledge(Message $message): void
     {
         $this->pubSub->acknowledge($message);
     }
+
+    /**
+     * Pull a specific PubSub message from PubSub topic specified by the given message id.
+     *
+     * @param Message $original
+     * @return Message
+     * @throws ErrorException
+     */
+    public function pullFreshMessage(Message $original): Message
+    {
+        $topic = $this->pubSub->getTopic($this->pubSub->getQueue(null));
+        $subscription = $topic->subscription($this->pubSub->getSubscriberName());
+
+        $messages = $subscription->pull(['returnImmediately' => true, 'maxMessages' => 10]);
+
+        foreach ($messages as $message) {
+            if (!($message instanceof Message)) {
+                continue;
+            }
+            if ($message->id() === $original->id()) {
+                return $message;
+            }
+        }
+
+        throw new ErrorException(sprintf('Failed to pull a PubSub message with id "%s"', $original->id()));
+    }
 }

src/GcrPubSubQueueFake.php

@@ -63,4 +63,15 @@ public function assertMessageHasAcknowledged(string $messageId): void
     {
         PHPUnit::assertTrue($this->acknowledgedMessages->has($messageId));
     }
+
+    /**
+     * Pull a specific PubSub message from PubSub topic specified by the given message id.
+     *
+     * @param Message $original
+     * @return Message
+     */
+    public function pullFreshMessage(Message $original): Message
+    {
+        return $original;
+    }
 }

src/Requests/PubSubEventRequest.php

@@ -4,6 +4,7 @@
 
 use Google\Cloud\PubSub\Message;
 use Illuminate\Foundation\Http\FormRequest;
+use RichanFongdasen\GCRWorker\Facade\GcrQueue;
 
 class PubSubEventRequest extends FormRequest
 {
@@ -27,7 +28,9 @@ public function getPubSubMessage(): Message
         $requestData = $this->all();
         $requestData['message']['data'] = base64_decode($requestData['message']['data']);
 
-        return new Message($requestData['message']);
+        $message = new Message($requestData['message']);
+
+        return GcrQueue::pullFreshMessage($message);
     }
 
     /**