[redis][poc] Add delay strategy

pkg/redis-tools/DelayStrategy.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace Enqueue\RedisTools;
+
+use Enqueue\Redis\RedisContext;
+use Enqueue\Redis\RedisDestination;
+use Enqueue\Redis\RedisMessage;
+
+interface DelayStrategy
+{
+    /**
+     * @param RedisContext     $context
+     * @param RedisDestination $dest
+     * @param RedisMessage     $message
+     * @param int              $delayMsec
+     */
+    public function delayMessage(RedisContext $context, RedisDestination $dest, RedisMessage $message, $delayMsec);
+
+    /**
+     * Search for delayed message and move them on main queue.
+     *
+     * @param RedisContext     $context
+     * @param RedisDestination $dest
+     */
+    public function processDelayedMessage(RedisContext $context, RedisDestination $dest);
+}

pkg/redis-tools/DelayStrategyAware.php

@@ -0,0 +1,13 @@
+<?php
+
+namespace Enqueue\RedisTools;
+
+interface DelayStrategyAware
+{
+    /**
+     * @param DelayStrategy $delayStrategy
+     *
+     * @return self
+     */
+    public function setDelayStrategy(DelayStrategy $delayStrategy = null);
+}

pkg/redis-tools/DelayStrategyAwareTrait.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace Enqueue\RedisTools;
+
+trait DelayStrategyAwareTrait
+{
+    /**
+     * @var DelayStrategy
+     */
+    protected $delayStrategy;
+
+    /**
+     * @param DelayStrategy|null $delayStrategy
+     *
+     * @return self
+     */
+    public function setDelayStrategy(DelayStrategy $delayStrategy = null)
+    {
+        $this->delayStrategy = $delayStrategy;
+
+        return $this;
+    }
+}

pkg/redis-tools/RedisZSetDelayConsumer.php

@@ -0,0 +1,136 @@
+<?php
+
+namespace Enqueue\RedisTools;
+
+use Enqueue\Redis\Redis;
+use Enqueue\Redis\RedisContext;
+use Enqueue\Redis\RedisDestination;
+use Enqueue\Redis\RedisMessage;
+use Interop\Queue\InvalidMessageException;
+use Interop\Queue\PsrConsumer;
+use Interop\Queue\PsrMessage;
+
+class RedisZSetDelayConsumer implements PsrConsumer
+{
+    /**
+     * @var RedisDestination
+     */
+    private $queue;
+
+    /**
+     * @var RedisContext
+     */
+    private $context;
+
+    /**
+     * @param RedisContext     $context
+     * @param RedisDestination $queue
+     */
+    public function __construct(RedisContext $context, RedisDestination $queue)
+    {
+        $this->context = $context;
+        $this->queue = $queue;
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return RedisDestination
+     */
+    public function getQueue()
+    {
+        return $this->queue;
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return RedisMessage|null
+     */
+    public function receive($timeout = 0)
+    {
+        return $this->receiveNoWait();
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return RedisMessage|null
+     */
+    public function receiveNoWait()
+    {
+        while (false !== ($timestamp = $this->nextDelayedTimestamp())) {
+            $this->enqueueDelayedMessagesForTimestamp($timestamp);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @param RedisMessage $message
+     */
+    public function acknowledge(PsrMessage $message)
+    {
+        // do nothing. redis transport always works in auto ack mode
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @param RedisMessage $message
+     */
+    public function reject(PsrMessage $message, $requeue = false)
+    {
+        InvalidMessageException::assertMessageInstanceOf($message, RedisMessage::class);
+
+        // do nothing on reject. redis transport always works in auto ack mode
+
+        if ($requeue) {
+            $this->context->createProducer()->send($this->queue, $message);
+        }
+    }
+
+    /**
+     * @return Redis
+     */
+    private function getRedis()
+    {
+        return $this->context->getRedis();
+    }
+
+    //TODO: refactor into a php generator
+    private function nextDelayedTimestamp()
+    {
+        $at = time();
+
+        //TODO:check zrange by score definition
+        $items = $this->getRedis()->zrangebyscore('enqueue:'.$this->getQueue()->getTopicName().':delayed', '-inf', $at, 'LIMIT', 0, 1);
+
+        if (!empty($items)) {
+            return $items[0];
+        }
+
+        return false;
+    }
+
+    private function enqueueDelayedMessagesForTimestamp($timestamp)
+    {
+        $message = null;
+        while ($message = $this->nextMessageForTimestamp($timestamp)) {
+            $this->context->createProducer()->send($this->queue, $message);
+        }
+    }
+
+    private function nextMessageForTimestamp($timestamp)
+    {
+        $queue = 'enqueue:'.$this->getQueue()->getTopicName().':delayed:'.$timestamp;
+        if ($message = $this->getRedis()->rpop($queue)) {
+            if (0 == $this->getRedis()->llen($queue)) {
+                $this->getRedis()->del($queue);
+                $this->getRedis()->zrem('enqueue:'.$this->getQueue()->getTopicName().':delayed', $timestamp);
+            }
+
+            return RedisMessage::jsonUnserialize($message);
+        }
+    }
+}

pkg/redis-tools/RedisZSetDelayStrategy.php

@@ -0,0 +1,43 @@
+<?php
+
+namespace Enqueue\RedisTools;
+
+use Enqueue\Redis\RedisContext;
+use Enqueue\Redis\RedisDestination;
+use Enqueue\Redis\RedisMessage;
+
+class RedisZSetDelayStrategy implements DelayStrategy
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function delayMessage(RedisContext $context, RedisDestination $dest, RedisMessage $message, $delayMsec)
+    {
+        $delayMessage = $context->createMessage($message->getBody(), $message->getProperties(), $message->getHeaders());
+        $delayMessage->setProperty('x-delay', (int) $delayMsec);
+
+        $targetTimestamp = time() + $delayMsec / 1000;
+
+        $context->getRedis()->zadd('enqueue:'.$dest->getTopicName().':delayed', $targetTimestamp, $targetTimestamp);
+
+        $delayTopic = $context->createTopic('enqueue:'.$dest->getTopicName().':delayed:'.$targetTimestamp);
+
+        $producer = $context->createProducer();
+
+        if ($producer instanceof DelayStrategyAware) {
+            $producer->setDelayStrategy(null);
+        }
+
+        $producer->send($delayTopic, $delayMessage);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function processDelayedMessage(RedisContext $context, RedisDestination $dest)
+    {
+        $delayConsumer = new RedisZSetDelayConsumer($context, $dest);
+
+        $delayConsumer->receive();
+    }
+}

pkg/redis-tools/composer.json

@@ -0,0 +1,36 @@
+{
+    "name": "enqueue/redis-tools",
+    "type": "library",
+    "description": "Message Queue Redis Tools",
+    "keywords": ["messaging", "queue", "redis"],
+    "homepage": "https://enqueue.forma-pro.com/",
+    "license": "MIT",
+    "require": {
+        "php": ">=5.6",
+        "queue-interop/queue-interop": "^0.6@dev|^1.0.0-alpha1"
+    },
+    "require-dev": {
+        "phpunit/phpunit": "~5.4.0",
+        "enqueue/test": "^0.8@dev",
+        "enqueue/null": "^0.8@dev"
+    },
+    "support": {
+        "email": "opensource@forma-pro.com",
+        "issues": "https://github.com/php-enqueue/enqueue-dev/issues",
+        "forum": "https://gitter.im/php-enqueue/Lobby",
+        "source": "https://github.com/php-enqueue/enqueue-dev",
+        "docs": "https://github.com/php-enqueue/enqueue-dev/blob/master/docs/index.md"
+    },
+    "autoload": {
+        "psr-4": { "Enqueue\\RedisTools\\": "" },
+        "exclude-from-classmap": [
+            "/Tests/"
+        ]
+    },
+    "minimum-stability": "dev",
+    "extra": {
+        "branch-alias": {
+            "dev-master": "0.8.x-dev"
+        }
+    }
+}

pkg/redis/Redis.php

@@ -35,4 +35,16 @@ public function disconnect();
      * @param string $key
      */
     public function del($key);
+
+    public function zrangebyscore($key, $min, $max, $offset, $limit, $options = []);
+
+    public function zadd($key, $score, $member);
+
+    public function rpush($key, $value);
+
+    public function lpop($key);
+
+    public function llen($key);
+
+    public function zrem($key, $member);
 }

pkg/redis/RedisConsumer.php

@@ -2,12 +2,15 @@
 
 namespace Enqueue\Redis;
 
+use Enqueue\RedisTools\DelayStrategyAware;
+use Enqueue\RedisTools\DelayStrategyAwareTrait;
 use Interop\Queue\InvalidMessageException;
 use Interop\Queue\PsrConsumer;
 use Interop\Queue\PsrMessage;
 
-class RedisConsumer implements PsrConsumer
+class RedisConsumer implements PsrConsumer, DelayStrategyAware
 {
+    use DelayStrategyAwareTrait;
     /**
      * @var RedisDestination
      */
@@ -45,6 +48,10 @@ public function getQueue()
      */
     public function receive($timeout = 0)
     {
+        if (null != $this->delayStrategy) {
+            $this->delayStrategy->processDelayedMessage($this->context, $this->queue);
+        }
+
         $timeout = (int) ($timeout / 1000);
         if (empty($timeout)) {
             //            Caused by
@@ -66,6 +73,10 @@ public function receive($timeout = 0)
      */
     public function receiveNoWait()
     {
+        if (null != $this->delayStrategy) {
+            $this->delayStrategy->processDelayedMessage($this->context, $this->queue);
+        }
+
         if ($message = $this->getRedis()->rpop($this->queue->getName())) {
             return RedisMessage::jsonUnserialize($message);
         }