Lines 100.00% 41 / 41
Functions and Methods 100.00% 5 / 5
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
WorkerMessageReader 100.00% 41 / 41 100.00% 5 / 5 14 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 readMessagesInBackground 100.00% 15 / 15 100.00% 1 / 1 4
 reportGone 100.00% 7 / 7 100.00% 1 / 1 2
 deliverToHandler 100.00% 7 / 7 100.00% 1 / 1 2
 receiveNextDecodableMessage 100.00% 11 / 11 100.00% 1 / 1 5
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Closure;
8use Psr\Log\LoggerInterface;
9use Stewart\Runtime\Exception\TransportError;
10use Stewart\Runtime\Exception\TransportException;
11use Stewart\Runtime\Ipc\Message\WorkerMessage;
12use Throwable;
13
14use function Amp\async;
15
16final readonly class WorkerMessageReader
17{
18    public function __construct(
19        private LoggerInterface $logger,
20    ) {}
21
22    /**
23     * @param Closure(WorkerMessage): void $onMessage
24     * @param Closure(string): void $onGone
25     */
26    public function readMessagesInBackground(WorkerHandle $handle, Closure $onMessage, Closure $onGone): void
27    {
28        async(function () use ($handle, $onMessage, $onGone): void {
29            try {
30                while (($message = $this->receiveNextDecodableMessage($handle)) !== null) {
31                    if ($message instanceof WorkerMessage) {
32                        $this->deliverToHandler($handle, $message, $onMessage);
33                    } else {
34                        $this->logger->warning('Dropped a message that is not a worker message', [
35                            'worker' => $handle->id->value,
36                            'message' => $message::class,
37                        ]);
38                    }
39                }
40
41                $reason = 'closed the channel';
42                $this->logger->debug('Stopped reading from worker', ['worker' => $handle->id->value, 'reason' => $reason]);
43            } catch (Throwable $e) {
44                $reason = $e->getMessage();
45                $this->logger->debug('Stopped reading from worker', ['worker' => $handle->id->value, 'exception' => $e]);
46            }
47
48            $this->reportGone($handle, $reason, $onGone);
49        })->ignore();
50    }
51
52    /** @param Closure(string): void $onGone */
53    private function reportGone(WorkerHandle $handle, string $reason, Closure $onGone): void
54    {
55        try {
56            $onGone($reason);
57        } catch (Throwable $e) {
58            $this->logger->error('Failed handling a worker exit', [
59                'worker' => $handle->id->value,
60                'reason' => $reason,
61                'exception' => $e,
62            ]);
63        }
64    }
65
66    /** @param Closure(WorkerMessage): void $onMessage */
67    private function deliverToHandler(WorkerHandle $handle, WorkerMessage $message, Closure $onMessage): void
68    {
69        try {
70            $onMessage($message);
71        } catch (Throwable $e) {
72            $this->logger->error('Failed handling a worker message', [
73                'worker' => $handle->id->value,
74                'message' => $message::class,
75                'exception' => $e,
76            ]);
77        }
78    }
79
80    /** @throws TransportException */
81    private function receiveNextDecodableMessage(WorkerHandle $handle): ?object
82    {
83        while (true) {
84            try {
85                return $handle->getTransport()->receive();
86            } catch (TransportException $e) {
87                // amphp closes the message channel on every worker exit; the exit result arrives through join().
88                if ($e->reason === TransportError::PeerDisconnected) {
89                    return null;
90                }
91
92                if ($e->reason !== TransportError::UndecodableFrame) {
93                    throw $e;
94                }
95
96                $this->logger->error('Dropped a worker message the broker could not decode', [
97                    'worker' => $handle->id->value,
98                    'exception' => $e,
99                ]);
100            }
101        }
102    }
103}