Added support for delay message

munir131 · Oct 7, 2019 · e7847ea · e7847ea
1 parent 04fedb1
commit e7847ea
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 0 deletions.
diff --git a/src/PubSubQueue.php b/src/PubSubQueue.php
@@ -165,6 +165,7 @@ public function pop($subscriber = null)
             'returnImmediately' => true,
             'maxMessages' => 1,
         ]);
+        $messages = $this->checkAvailabilty($messages);
 
         if (!empty($messages) && count($messages) > 0) {
             return new PubSubJob(
@@ -178,6 +179,18 @@ public function pop($subscriber = null)
         }
     }
 
+    public function checkAvailabilty($messages) {
+        if (!empty($messages) && count($messages) > 0) {
+            $availableAt = $messages[0]->attribute('available_at');
+            $now = (new \DateTime('now'))->getTimestamp();
+            if (!$availableAt) {
+                return $messages;
+            } else if ($availableAt < $now) {
+                return $messages;
+            }
+        }
+    }
+
     /**
      * Push an array of jobs onto the queue.
      *

diff --git a/tests/Unit/PubSubQueueTests.php b/tests/Unit/PubSubQueueTests.php
@@ -298,4 +298,30 @@ public function testGetPubSub()
     {
         $this->assertTrue($this->queue->getPubSub() instanceof PubSubClient);
     }
+
+    public function testDelayedMessage()
+    {
+        $this->subscription->method('pull')
+            ->willReturn([$this->message]);
+
+        $this->message->method('attribute')
+            ->with($this->equalTo('available_at'))
+            ->willReturn((new \DateTime('now +1seconds'))->getTimestamp());
+
+        $this->topic->method('subscription')
+            ->willReturn($this->subscription);
+
+        $this->topic->method('exists')
+            ->willReturn(true);
+
+        $this->queue->method('getTopic')
+            ->willReturn($this->topic);
+
+        $this->queue->setContainer($this->createMock(Container::class));
+
+        $this->assertTrue(is_null($this->queue->pop('test')));
+        sleep(2);
+        $this->assertTrue($this->queue->pop('test') instanceof PubSubJob);
+
+    }
 }