Lines 96.55% 56 / 58
Methods 92.85% 13 / 14
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __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
 [Tomaj\Hermes\Driver\MaxItemsTrait] setMaxProcessItems 100.00% 1 / 1 100.00% 1 / 1 1
 [Tomaj\Hermes\Driver\MaxItemsTrait] incrementProcessedItems 100.00% 2 / 2 100.00% 1 / 1 1
 [Tomaj\Hermes\Driver\MaxItemsTrait] processed 100.00% 1 / 1 100.00% 1 / 1 1
 [Tomaj\Hermes\Driver\MaxItemsTrait] shouldProcessNext 100.00% 3 / 3 100.00% 1 / 1 2
 [Tomaj\Hermes\Driver\ShutdownTrait] setShutdown 100.00% 2 / 2 100.00% 1 / 1 1
 [Tomaj\Hermes\Driver\ShutdownTrait] shouldShutdown 100.00% 5 / 5 100.00% 1 / 1 4
 [Tomaj\Hermes\Driver\ShutdownTrait] checkShutdown 100.00% 2 / 2 100.00% 1 / 1 2
 [Tomaj\Hermes\Driver\SerializerAwareTrait] setSerializer 100.00% 1 / 1 100.00% 1 / 1 1
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}

From Tomaj\Hermes\Driver\MaxItemsTrait

7trait MaxItemsTrait
8{
9    private int $processed = 0;
10
11    private int $maxProcessItems = 0;
12
13    public function setMaxProcessItems(int $count): void
14    {
15        $this->maxProcessItems = $count;
16    }
17
18    public function incrementProcessedItems(): int
19    {
20        $this->processed++;
21        return $this->processed;
22    }
23
24    public function processed(): int
25    {
26        return $this->processed;
27    }
28
29    public function shouldProcessNext(): bool
30    {
31        if ($this->maxProcessItems === 0) {
32            return true;
33        }
34        return $this->processed < $this->maxProcessItems;
35    }
36}

From Tomaj\Hermes\Driver\ShutdownTrait

12trait ShutdownTrait
13{
14    private ?ShutdownInterface $shutdown = null;
15
16    private ?DateTimeImmutable $startTime = null;
17
18    public function setShutdown(ShutdownInterface $shutdown): void
19    {
20        $this->shutdown = $shutdown;
21        $this->startTime = new DateTimeImmutable();
22    }
23
24    private function shouldShutdown(): bool
25    {
26        if (SignalStop::received() !== null) {
27            return true;
28        }
29        if ($this->shutdown === null || $this->startTime === null) {
30            return false;
31        }
32        return $this->shutdown->shouldShutdown($this->startTime);
33    }
34
35    /**
36     * @throws ShutdownException
37     */
38    private function checkShutdown(): void
39    {
40        if ($this->shouldShutdown()) {
41            throw new ShutdownException();
42        }
43    }
44}

From Tomaj\Hermes\Driver\SerializerAwareTrait

9trait SerializerAwareTrait
10{
11    private SerializerInterface $serializer;
12
13    /**
14     * Set serializer to driver
15     *
16     * You can this trait to set serializer from outsite to your driver
17     * if you need your custom serialization for your objects.
18     */
19    public function setSerializer(SerializerInterface $serializer): void
20    {
21        $this->serializer = $serializer;
22    }
23}