Lines 97.91% 47 / 48
Functions and Methods 83.33% 5 / 6
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
LazyRabbitMqDriver 97.91% 47 / 48 83.33% 5 / 6 16 0.00% 0 / 1
 __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
1<?php
2
3declare(strict_types=1);
4
5namespace Tomaj\Hermes\Driver;
6
7use Closure;
8use PhpAmqpLib\Channel\AMQPChannel;
9use PhpAmqpLib\Connection\AbstractConnection;
10use PhpAmqpLib\Exception\AMQPChannelClosedException;
11use PhpAmqpLib\Exception\AMQPConnectionBlockedException;
12use PhpAmqpLib\Exception\AMQPConnectionClosedException;
13use PhpAmqpLib\Exception\AMQPOutOfBoundsException;
14use PhpAmqpLib\Exception\AMQPRuntimeException;
15use PhpAmqpLib\Exception\AMQPTimeoutException;
16use PhpAmqpLib\Message\AMQPMessage;
17use Throwable;
18use Tomaj\Hermes\Dispatcher;
19use Tomaj\Hermes\MessageInterface;
20use Tomaj\Hermes\MessageSerializer;
21use Tomaj\Hermes\SerializeException;
22use Tomaj\Hermes\Shutdown\ShutdownException;
23
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}