Lines 97.82% 45 / 46
Functions and Methods 80.00% 4 / 5
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
AmazonSqsDriver 97.82% 45 / 46 80.00% 4 / 5 16 0.00% 0 / 1
 __construct 92.30% 12 / 13 0.00% 0 / 1 3.00
 send 100.00% 5 / 5 100.00% 1 / 1 1
 setupPriorityQueue 100.00% 1 / 1 100.00% 1 / 1 1
 wait 100.00% 23 / 23 100.00% 1 / 1 10
 deleteMessage 100.00% 4 / 4 100.00% 1 / 1 1
1<?php
2
3declare(strict_types=1);
4
5namespace Tomaj\Hermes\Driver;
6
7use Aws\Sqs\SqsClient;
8use Closure;
9use Tomaj\Hermes\Dispatcher;
10use Tomaj\Hermes\MessageInterface;
11use Tomaj\Hermes\MessageSerializer;
12use Tomaj\Hermes\Shutdown\ShutdownException;
13
14final class AmazonSqsDriver implements DriverInterface
15{
16    use MaxItemsTrait;
17    use ShutdownTrait;
18    use SerializerAwareTrait;
19
20    private readonly string $queueUrl;
21
22    /**
23     * Create new Amazon SQS driver.
24     *
25     * You have to create aws client instnace and provide it to this driver.
26     * You can use service builder or factory method.
27     *
28     * <code>
29     *  use Aws\Sqs\SqsClient;
30     *
31     *  $client = SqsClient::factory(array(
32     *    'profile' => '<profile in your aws credentials file>',
33     *    'region'  => '<region name>'
34     *  ));
35     * </code>
36     *
37     * or
38     *
39     * <code>
40     * use Aws\Common\Aws;
41     *
42     * // Create a service builder using a configuration file
43     * $aws = Aws::factory('/path/to/my_config.json');
44     *
45     * // Get the client from the builder by namespace
46     * $client = $aws->get('Sqs');
47     * </code>
48     *
49     * More examples see: https://docs.aws.amazon.com/aws-sdk-php/v2/guide/service-sqs.html
50     *
51     * @see examples/sqs folder
52     *
53     * @param array<string, scalar|null> $queueAttributes
54     */
55    public function __construct(
56        private readonly SqsClient $client,
57        private readonly string $queueName,
58        array $queueAttributes = [],
59    ) {
60        $this->serializer = new MessageSerializer();
61
62        $attributes = array_map(
63            static fn(mixed $attributeValue): string => is_scalar($attributeValue) ? (string) $attributeValue : '',
64            $queueAttributes,
65        );
66
67        $result = $this->client->createQueue([
68            'QueueName' => $this->queueName,
69            'Attributes' => $attributes,
70        ]);
71
72        $queueUrl = $result->get('QueueUrl');
73        if (!is_string($queueUrl)) {
74            throw new \RuntimeException('Amazon SQS createQueue response did not include a queue URL');
75        }
76        $this->queueUrl = $queueUrl;
77    }
78
79    #[\Override]
80    public function send(MessageInterface $message, int $priority = Dispatcher::DEFAULT_PRIORITY): bool
81    {
82        $this->client->sendMessage([
83            'QueueUrl' => $this->queueUrl,
84            'MessageBody' => $this->serializer->serialize($message),
85        ]);
86        return true;
87    }
88
89    /**
90     * @throws NotSupportedException
91     */
92    #[\Override]
93    public function setupPriorityQueue(string $name, int $priority): never
94    {
95        throw new NotSupportedException('AmazonSQS is not supporting priority queues now');
96    }
97
98    #[\Override]
99    public function wait(Closure $callback, array $priorities = []): void
100    {
101        while (true) {
102            $this->checkShutdown();
103            if (!$this->shouldProcessNext()) {
104                break;
105            }
106
107            $result = $this->client->receiveMessage([
108                'QueueUrl' => $this->queueUrl,
109                'WaitTimeSeconds' => 1,
110            ]);
111
112            $messages = $result['Messages'] ?? null;
113
114            if (is_array($messages) && $messages !== []) {
115                foreach ($messages as $message) {
116                    if (!is_array($message)) {
117                        continue;
118                    }
119                    $receiptHandle = $message['ReceiptHandle'] ?? null;
120                    $body = $message['Body'] ?? null;
121                    if (!is_string($receiptHandle) || !is_string($body)) {
122                        continue;
123                    }
124                    // Delete only after the message was handled. If unserializing or the callback fails,
125                    // the message stays in SQS and is redelivered after the visibility timeout (or moved to a DLQ).
126                    try {
127                        $callback($this->serializer->unserialize($body));
128                    } catch (ShutdownException $e) {
129                        // The dispatcher requests shutdown after the message was handled.
130                        $this->deleteMessage($receiptHandle);
131                        throw $e;
132                    }
133                    $this->deleteMessage($receiptHandle);
134                    $this->incrementProcessedItems();
135                }
136            }
137        }
138    }
139
140    private function deleteMessage(string $receiptHandle): void
141    {
142        $this->client->deleteMessage([
143            'QueueUrl' => $this->queueUrl,
144            'ReceiptHandle' => $receiptHandle,
145        ]);
146    }
147}