Lines 95.12% 39 / 41
Functions and Methods 83.33% 5 / 6
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
RedisSetDriver 95.12% 39 / 41 83.33% 5 / 6 21 0.00% 0 / 1
 __construct 100.00% 2 / 2 100.00% 1 / 1 1
 send 100.00% 5 / 5 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 92.30% 24 / 26 0.00% 0 / 1 11.06
 pop 100.00% 4 / 4 100.00% 1 / 1 3
1<?php
2
3declare(strict_types=1);
4
5namespace Tomaj\Hermes\Driver;
6
7use Closure;
8use Redis;
9use Tomaj\Hermes\Dispatcher;
10use Tomaj\Hermes\MessageInterface;
11use Tomaj\Hermes\MessageSerializer;
12use Tomaj\Hermes\SerializeException;
13use Tomaj\Hermes\Shutdown\ShutdownException;
14
15final class RedisSetDriver 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 RedisSetDriver
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 have to have enabled native Redis php extension.
33     *
34     * @see examples/redis
35     */
36    public function __construct(
37        private readonly Redis $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     */
49    #[\Override]
50    public function send(MessageInterface $message, int $priority = Dispatcher::DEFAULT_PRIORITY): bool
51    {
52        if ($message->getExecuteAt() !== null && $message->getExecuteAt() > microtime(true)) {
53            $this->redis->zAdd($this->scheduleKey, $message->getExecuteAt(), $this->serializer->serialize($message));
54        } else {
55            $key = $this->getKey($priority);
56            $this->redis->sAdd($key, $this->serializer->serialize($message));
57        }
58        return true;
59    }
60
61    #[\Override]
62    public function setupPriorityQueue(string $name, int $priority): void
63    {
64        $this->queues[$priority] = $name;
65    }
66
67    /**
68     * @throws UnknownPriorityException
69     */
70    private function getKey(int $priority): string
71    {
72        if (!isset($this->queues[$priority])) {
73            throw new UnknownPriorityException("Unknown priority {$priority}");
74        }
75        return $this->queues[$priority];
76    }
77
78    /**
79     * @throws ShutdownException
80     * @throws UnknownPriorityException
81     * @throws SerializeException
82     */
83    #[\Override]
84    public function wait(Closure $callback, array $priorities = []): void
85    {
86        // Highest priority first, regardless of the order the queues were set up in
87        $queues = $this->queues;
88        krsort($queues, SORT_NUMERIC);
89        while (true) {
90            $this->checkShutdown();
91            if (!$this->shouldProcessNext()) {
92                break;
93            }
94
95            // check schedule
96            /** @var list<string> $messageStrings */
97            $messageStrings = $this->redis->zRangeByScore($this->scheduleKey, '-inf', (string) microtime(true), ['limit' => [0, 1]]);
98            foreach ($messageStrings as $messageString) {
99                // ZREM is atomic: only the worker that actually removed the entry re-sends it
100                if ($this->redis->zRem($this->scheduleKey, $messageString) === 1) {
101                    $this->send($this->serializer->unserialize($messageString));
102                }
103            }
104
105            $messageString = null;
106            $foundPriority = null;
107
108            foreach ($queues as $priority => $name) {
109                if ($priorities !== [] && !in_array($priority, $priorities, true)) {
110                    continue;
111                }
112                if ($messageString !== null) {
113                    break;
114                }
115
116                $messageString = $this->pop($this->getKey($priority));
117                $foundPriority = $priority;
118            }
119
120            if ($messageString !== null) {
121                $message = $this->serializer->unserialize($messageString);
122                $callback($message, $foundPriority);
123                $this->incrementProcessedItems();
124            } else {
125                if ($this->refreshInterval !== 0) {
126                    $this->checkShutdown();
127                    sleep($this->refreshInterval);
128                }
129            }
130        }
131    }
132
133    private function pop(string $key): ?string
134    {
135        $messageString = $this->redis->sPop($key);
136        if (is_string($messageString) && $messageString !== '') {
137            return $messageString;
138        }
139
140        return null;
141    }
142}