Lines 97.77% 44 / 45
Functions and Methods 80.00% 4 / 5
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
PredisSetDriver 97.77% 44 / 45 80.00% 4 / 5 20 0.00% 0 / 1
 __construct 100.00% 2 / 2 100.00% 1 / 1 1
 send 100.00% 6 / 6 100.00% 1 / 1 3
 setupPriorityQueue 100.00% 1 / 1 100.00% 1 / 1 1
 getKey 100.00% 3 / 3 100.00% 1 / 1 2
 wait 96.96% 32 / 33 0.00% 0 / 1 13
1<?php
2
3declare(strict_types=1);
4
5namespace Tomaj\Hermes\Driver;
6
7use Closure;
8use Predis\Client;
9use Tomaj\Hermes\Dispatcher;
10use Tomaj\Hermes\MessageInterface;
11use Tomaj\Hermes\MessageSerializer;
12use Tomaj\Hermes\SerializeException;
13use Tomaj\Hermes\Shutdown\ShutdownException;
14
15final class PredisSetDriver implements DriverInterface
16{
17    use MaxItemsTrait;
18    use ShutdownTrait;
19    use SerializerAwareTrait;
20
21    /** @var array<int, string> */
22    private array $queues = [];
23
24    /**
25     * Create new PredisSetDriver
26     *
27     * This driver is using redis set. With send message it add new item to set
28     * and in wait() command it is reading new items in this set.
29     * This driver doesn't use redis pubsub functionality, only redis sets.
30     *
31     * Managing connection to redis is up to you and you have to create it outsite
32     * of this class. You will need to install predis php package.
33     *
34     * @see examples/redis
35     */
36    public function __construct(
37        private readonly Client $redis,
38        string $key = 'hermes',
39        private readonly int $refreshInterval = 1,
40        private readonly string $scheduleKey = 'hermes_schedule',
41    ) {
42        $this->setupPriorityQueue($key, Dispatcher::DEFAULT_PRIORITY);
43        $this->serializer = new MessageSerializer();
44    }
45
46    /**
47     * @throws UnknownPriorityException
48     * @throws SerializeException
49     */
50    #[\Override]
51    public function send(MessageInterface $message, int $priority = Dispatcher::DEFAULT_PRIORITY): bool
52    {
53        $executeAt = $message->getExecuteAt();
54        if ($executeAt !== null && $executeAt > microtime(true)) {
55            $this->redis->zadd($this->scheduleKey, [$this->serializer->serialize($message) => $executeAt]);
56        } else {
57            $key = $this->getKey($priority);
58            $this->redis->sadd($key, [$this->serializer->serialize($message)]);
59        }
60        return true;
61    }
62
63    #[\Override]
64    public function setupPriorityQueue(string $name, int $priority): void
65    {
66        $this->queues[$priority] = $name;
67    }
68
69    /**
70     * @throws UnknownPriorityException
71     */
72    private function getKey(int $priority): string
73    {
74        if (!isset($this->queues[$priority])) {
75            throw new UnknownPriorityException("Unknown priority {$priority}");
76        }
77        return $this->queues[$priority];
78    }
79
80    /**
81     * @throws ShutdownException
82     * @throws UnknownPriorityException
83     * @throws SerializeException
84     */
85    #[\Override]
86    public function wait(Closure $callback, array $priorities = []): void
87    {
88        // Highest priority first, regardless of the order the queues were set up in
89        $queues = $this->queues;
90        krsort($queues, SORT_NUMERIC);
91        while (true) {
92            $this->checkShutdown();
93            if (!$this->shouldProcessNext()) {
94                break;
95            }
96
97            // check schedule; Redis takes the score bound as a string, keep the microseconds
98            $now = sprintf('%.6F', microtime(true));
99            $messagesString = $this->redis->zrangebyscore($this->scheduleKey, '-inf', $now, ['LIMIT' => ['OFFSET' => 0, 'COUNT' => 1]]);
100            if ($messagesString !== []) {
101                foreach ($messagesString as $messageString) {
102                    if (!is_string($messageString)) {
103                        continue;
104                    }
105                    $this->redis->zrem($this->scheduleKey, $messageString);
106                    $this->send($this->serializer->unserialize($messageString));
107                }
108            }
109
110            $messageString = null;
111            $foundPriority = null;
112
113            foreach ($queues as $priority => $name) {
114                if ($priorities !== [] && !in_array($priority, $priorities, true)) {
115                    continue;
116                }
117                if ($messageString !== null) {
118                    break;
119                }
120
121                $key = $this->getKey($priority);
122
123                $popped = $this->redis->spop($key);
124                if (is_string($popped)) {
125                    $messageString = $popped;
126                    $foundPriority = $priority;
127                }
128            }
129
130            if (is_string($messageString)) {
131                $message = $this->serializer->unserialize($messageString);
132                $callback($message, $foundPriority);
133                $this->incrementProcessedItems();
134            } else {
135                if ($this->refreshInterval !== 0) {
136                    $this->checkShutdown();
137                    sleep($this->refreshInterval);
138                }
139            }
140        }
141    }
142}