Lines 98.46% 64 / 65
Methods 92.85% 13 / 14
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 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
 wait 95.45% 21 / 22 0.00% 0 / 1 5
 handleDelivery 100.00% 13 / 13 100.00% 1 / 1 4
 getChannel 100.00% 5 / 5 100.00% 1 / 1 2
 [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
24final class LazyRabbitMqDriver implements DriverInterface
25{
26    use MaxItemsTrait;
27    use ShutdownTrait;
28    use SerializerAwareTrait;
29
30    private ?AMQPChannel $channel = null;
31
32    /**
33     * @param array<string, mixed> $amqpMessageProperties
34     * @param bool $durable Declare a durable queue and publish persistent messages (delivery_mode 2).
35     *                      Pass false to keep using a queue that was declared non-durable by Hermes 5.x.
36     *                      RabbitMQ 4.3+ refuses non-durable queues unless
37     *                      deprecated_features.permit.transient_nonexcl_queues is set.
38     */
39    public function __construct(
40        private readonly AbstractConnection $connection,
41        private readonly string $queue,
42        private readonly array $amqpMessageProperties = [],
43        private readonly int $refreshInterval = 0,
44        private readonly string $consumerTag = 'hermes',
45        private readonly bool $durable = true,
46    ) {
47        $this->serializer = new MessageSerializer();
48    }
49
50    /**
51     * @throws AMQPChannelClosedException
52     * @throws AMQPConnectionBlockedException
53     * @throws AMQPConnectionClosedException
54     * @throws AMQPTimeoutException
55     */
56    #[\Override]
57    public function send(MessageInterface $message, int $priority = Dispatcher::DEFAULT_PRIORITY): bool
58    {
59        $properties = $this->amqpMessageProperties;
60        if ($this->durable && !array_key_exists('delivery_mode', $properties)) {
61            $properties['delivery_mode'] = AMQPMessage::DELIVERY_MODE_PERSISTENT;
62        }
63        $rabbitMessage = new AMQPMessage($this->serializer->serialize($message), $properties);
64        $this->getChannel()->basic_publish($rabbitMessage, '', $this->queue);
65        return true;
66    }
67
68    /**
69     * @throws NotSupportedException
70     */
71    #[\Override]
72    public function setupPriorityQueue(string $name, int $priority): never
73    {
74        throw new NotSupportedException('LazyRabbitMqDriver is not supporting priority queues now');
75    }
76
77    /**
78     * @throws ShutdownException
79     * @throws AMQPOutOfBoundsException
80     * @throws AMQPRuntimeException
81     * @throws AMQPTimeoutException
82     * @throws \ErrorException
83     */
84    #[\Override]
85    public function wait(Closure $callback, array $priorities = []): void
86    {
87        // Deliver one unacknowledged message at a time to this worker.
88        $this->getChannel()->basic_qos(0, 1, false);
89        while (true) {
90            $this->getChannel()->basic_consume(
91                $this->queue,
92                $this->consumerTag,
93                false,
94                false,
95                false,
96                false,
97                function (AMQPMessage $rabbitMessage) use ($callback): void {
98                    $this->handleDelivery($rabbitMessage, $callback);
99                },
100            );
101
102            while ($this->getChannel()->is_consuming()) {
103                $this->getChannel()->wait(null, true);
104                $this->checkShutdown();
105                if (!$this->shouldProcessNext()) {
106                    break 2;
107                }
108                if ($this->refreshInterval !== 0) {
109                    sleep($this->refreshInterval);
110                }
111            }
112        }
113
114        $this->getChannel()->close();
115        $this->connection->close();
116    }
117
118    /**
119     * Acknowledge the delivery only after the callback finished.
120     *
121     * A message that cannot be unserialized is rejected without requeue (it goes to a dead-letter
122     * exchange when the queue has one). A failing callback requeues the message for another attempt.
123     *
124     * @throws SerializeException
125     * @throws ShutdownException
126     */
127    private function handleDelivery(AMQPMessage $rabbitMessage, Closure $callback): void
128    {
129        try {
130            $message = $this->serializer->unserialize($rabbitMessage->getBody());
131        } catch (SerializeException $e) {
132            $rabbitMessage->nack(false);
133            throw $e;
134        }
135
136        try {
137            $callback($message);
138        } catch (ShutdownException $e) {
139            // The dispatcher asks to stop after the message was handled.
140            $rabbitMessage->ack();
141            throw $e;
142        } catch (Throwable $e) {
143            $rabbitMessage->nack(true);
144            throw $e;
145        }
146        $rabbitMessage->ack();
147        $this->incrementProcessedItems();
148    }
149
150    /**
151     * @throws AMQPTimeoutException
152     */
153    private function getChannel(): AMQPChannel
154    {
155        if ($this->channel instanceof AMQPChannel) {
156            return $this->channel;
157        }
158        $this->channel = $this->connection->channel();
159        $this->channel->queue_declare($this->queue, false, $this->durable, false, false);
160        return $this->channel;
161    }
162}

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}