Lines 98.41% 62 / 63
Methods 92.30% 12 / 13
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __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
 [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
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}

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}