Lines 100.00% 27 / 27
Functions and Methods 100.00% 2 / 2
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
BrokerMessageReader 100.00% 27 / 27 100.00% 2 / 2 9 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 readUntilChannelCloses 100.00% 26 / 26 100.00% 1 / 1 8
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Worker;
6
7use Psr\Log\LoggerInterface;
8use Stewart\Runtime\Exception\TransportError;
9use Stewart\Runtime\Exception\TransportException;
10use Stewart\Runtime\Ipc\Message\BrokerMessage;
11use Stewart\Runtime\Ipc\Transport;
12use Stewart\Runtime\Worker\Message\BrokerMessageDispatcher;
13use Throwable;
14
15final readonly class BrokerMessageReader
16{
17    public function __construct(
18        private Transport $transport,
19        private BrokerMessageDispatcher $messages,
20        private WorkerShutdown $shutdown,
21        private LoggerInterface $logger,
22    ) {}
23
24    public function readUntilChannelCloses(): void
25    {
26        while (true) {
27            try {
28                $message = $this->transport->receive();
29            } catch (TransportException $e) {
30                if ($e->reason === TransportError::UndecodableFrame) {
31                    $this->logger->warning('Dropped a broker message the worker could not decode', [
32                        'exception' => $e,
33                    ]);
34
35                    continue;
36                }
37
38                $this->shutdown->stopAfterBrokerLoss('broker gone: ' . $e->getMessage());
39
40                return;
41            } catch (Throwable $e) {
42                $this->logger->error('Worker stopped reading broker messages', ['exception' => $e]);
43                $this->shutdown->stopAfterBrokerLoss('reading from the broker failed: ' . $e->getMessage());
44
45                return;
46            }
47
48            if ($message === null) {
49                $this->shutdown->stopAfterBrokerLoss('channel closed');
50
51                return;
52            }
53
54            if (!$message instanceof BrokerMessage) {
55                $this->logger->warning('Dropped a message that is not a broker message', ['message' => $message::class]);
56
57                continue;
58            }
59
60            try {
61                $this->messages->dispatch($message);
62            } catch (Throwable $e) {
63                $this->logger->error('Worker failed handling a broker message', [
64                    'message' => $message::class,
65                    'exception' => $e,
66                ]);
67            }
68        }
69    }
70}