Merge pull request #24395 from totten/master-release-time

Queue API - Expose option for delayed tasks (`$options['release_time']`)
civicrm · Aug 27, 2022 · 27d5300 · 27d5300
2 parents 8fdcb4d + 5891a2a
commit 27d5300
Show file tree

Hide file tree

Showing 4 changed files with 42 additions and 0 deletions.
diff --git a/CRM/Queue/Queue.php b/CRM/Queue/Queue.php
@@ -130,6 +130,7 @@ abstract public function existsQueue();
    * @param array $options
    *   Queue-dependent options; for example, if this is a
    *   priority-queue, then $options might specify the item's priority.
+   *   Ex: ['release_time' => strtotime('+3 hours')]
    */
   abstract public function createItem($data, $options = []);
 

diff --git a/CRM/Queue/Queue/Memory.php b/CRM/Queue/Queue/Memory.php
@@ -97,12 +97,16 @@ public function existsQueue() {
    * @param array $options
    *   Queue-dependent options; for example, if this is a
    *   priority-queue, then $options might specify the item's priority.
+   *   Ex: ['release_time' => strtotime('+3 hours')]
    */
   public function createItem($data, $options = []) {
     $id = $this->nextQueueItemId++;
     // force copy, no unintendedsharing effects from pointers
     $this->items[$id] = serialize($data);
     $this->runCounts[$id] = 0;
+    if (isset($options['release_time'])) {
+      $this->releaseTimes[$id] = $options['release_time'];
+    }
   }
 
   /**

diff --git a/CRM/Queue/Queue/SqlTrait.php b/CRM/Queue/Queue/SqlTrait.php
@@ -73,13 +73,17 @@ public function numberOfItems() {
    * @param array $options
    *   Queue-dependent options; for example, if this is a
    *   priority-queue, then $options might specify the item's priority.
+   *   Ex: ['release_time' => strtotime('+3 hours')]
    */
   public function createItem($data, $options = []) {
     $dao = new CRM_Queue_DAO_QueueItem();
     $dao->queue_name = $this->getName();
     $dao->submit_time = CRM_Utils_Time::getTime('YmdHis');
     $dao->data = serialize($data);
     $dao->weight = CRM_Utils_Array::value('weight', $options, 0);
+    if (isset($options['release_time'])) {
+      $dao->release_time = date('Y-m-d H:i:s', $options['release_time']);
+    }
     $dao->save();
   }
 

diff --git a/tests/phpunit/api/v4/Entity/QueueTest.php b/tests/phpunit/api/v4/Entity/QueueTest.php
@@ -211,6 +211,39 @@ public function testEmptyPoll() {
     $this->assertEquals(0, $startResult->count());
   }
 
+  public function getDelayableDrivers(): array {
+    return [
+      'Sql' => [['type' => 'Sql', 'runner' => 'task', 'error' => 'delete']],
+      'SqlParallel' => [['type' => 'SqlParallel', 'runner' => 'task', 'error' => 'delete']],
+      'Memory' => [['type' => 'Memory', 'runner' => 'task', 'error' => 'delete']],
+    ];
+  }
+
+  /**
+   * @dataProvider getDelayableDrivers
+   */
+  public function testDelayedStart(array $queueSpec) {
+    $queueName = 'QueueTest_' . md5(random_bytes(32)) . '_delayed';
+    $queue = \Civi::queue($queueName, $queueSpec);
+    $this->assertEquals(0, $queue->numberOfItems());
+
+    $releaseTime = \CRM_Utils_Time::strtotime('+3 seconds');
+    \Civi::queue($queueName)->createItem(new \CRM_Queue_Task(
+      [QueueTest::class, 'doSomething'],
+      ['itwillstartanymomentnow']
+    ), ['release_time' => $releaseTime]);
+    $this->assertEquals(1, $queue->numberOfItems());
+
+    // Not available... yet...
+    $claim1 = $queue->claimItem();
+    $this->assertEquals(NULL, $claim1);
+
+    // OK, it'll come in a few seconds...
+    $claim2 = $this->waitForClaim(0.5, 6, $queueName);
+    $this->assertEquals('itwillstartanymomentnow', $claim2['data']['arguments'][0]);
+    $this->assertTrue(\CRM_Utils_Time::time() >= $releaseTime);
+  }
+
   public function getErrorModes(): array {
     return [
       'delete' => ['delete'],