Lines 90.19% 46 / 51
Methods 80.00% 4 / 5
Classes 0.00% 0 / 1
Covered by tests of size
Name Lines Methods CRAP
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 registerHandler 100.00% 1 / 1 100.00% 1 / 1 1
 runOnce 100.00% 20 / 20 100.00% 1 / 1 5
 run 100.00% 14 / 14 100.00% 1 / 1 5
 decodeMessage 66.66% 10 / 15 0.00% 0 / 1 8.81
28class Worker
29{
30    private const int MAX_RETRIES = 3;
31
32    /** @var array<string, callable> */
33    private array $handlers = [];
34
35    public function __construct(
36        private readonly DatabaseTransport $databaseTransport,
37    ) {
38    }
39
40    public function registerHandler(string $messageClass, callable $handler): void
41    {
42        $this->handlers[$messageClass] = $handler;
43    }
44
45    /**
46     * Runs one available job.
47     */
48    public function runOnce(string $queue = 'default'): bool
49    {
50        $job = $this->databaseTransport->reserve($queue);
51        if ($job === null) {
52            return false;
53        }
54
55        $jobId = (int) $job['id'];
56        $headers = $job['headers'];
57
58        try {
59            $message = $this->decodeMessage($job['body']);
60            $handler = $this->handlers[$message::class] ?? null;
61
62            if (!is_callable($handler)) {
63                throw new RuntimeException('No queue handler registered for message class: ' . $message::class);
64            }
65
66            $handler($message);
67            $this->databaseTransport->acknowledge($jobId);
68
69            return true;
70        } catch (Throwable) {
71            $attempts = (int) ($headers['attempts'] ?? 0) + 1;
72            $headers['attempts'] = $attempts;
73
74            if ($attempts >= self::MAX_RETRIES) {
75                $this->databaseTransport->acknowledge($jobId);
76                return true;
77            }
78
79            $this->databaseTransport->release($jobId, new DateTimeImmutable('+60 seconds'), $headers);
80            return true;
81        }
82    }
83
84    /**
85     * Runs until the queue is empty or the maximum number of jobs has been processed.
86     */
87    public function run(int $maxJobs = 0, string $queue = 'default'): int
88    {
89        $processed = 0;
90
91        while ($maxJobs === 0 || $processed < $maxJobs) {
92            try {
93                if (!$this->runOnce($queue)) {
94                    break;
95                }
96
97                ++$processed;
98            } catch (Throwable $exception) {
99                error_log(sprintf(
100                    'Queue worker error in run() while processing queue "%s": %s in %s:%d',
101                    $queue,
102                    $exception->getMessage(),
103                    $exception->getFile(),
104                    $exception->getLine(),
105                ));
106            }
107        }
108
109        return $processed;
110    }
111
112    private function decodeMessage(string $body): QueueMessageInterface
113    {
114        $decoded = json_decode($body, associative: true);
115        if (!is_array($decoded) || !array_key_exists('class', $decoded)) {
116            throw new RuntimeException('Queue job body has an invalid format.');
117        }
118
119        $messageClass = (string) $decoded['class'];
120        $rawPayload = is_array($decoded['payload'] ?? null) ? $decoded['payload'] : [];
121        $payload = [];
122        foreach ($rawPayload as $payloadKey => $payloadValue) {
123            $payload[(string) $payloadKey] = $payloadValue;
124        }
125
126        if (!class_exists($messageClass)) {
127            throw new RuntimeException('Queue job references unknown message class: ' . $messageClass);
128        }
129
130        if (!is_subclass_of($messageClass, QueueMessageInterface::class)) {
131            throw new RuntimeException(
132                'Queue message class ' . $messageClass . ' does not implement ' . QueueMessageInterface::class,
133            );
134        }
135
136        /** @var class-string<QueueMessageInterface> $messageClass */
137        return $messageClass::fromArray($payload);
138    }
139}