Lines 89.47% 17 / 19
Functions and Methods 85.71% 6 / 7
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
ExposureRequester 89.47% 17 / 19 85.71% 6 / 7 10.12 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 requestExposure 100.00% 2 / 2 100.00% 1 / 1 1
 requestReconfiguration 0.00% 0 / 2 0.00% 0 / 1 2
 requestUpdate 100.00% 2 / 2 100.00% 1 / 1 1
 requestRemoval 100.00% 2 / 2 100.00% 1 / 1 1
 describeUnreachable 100.00% 1 / 1 100.00% 1 / 1 1
 awaitAnswer 100.00% 9 / 9 100.00% 1 / 1 4
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Worker\Exposure;
6
7use Amp\CancelledException;
8use Closure;
9use Stewart\Contracts\Exception\ExposureError;
10use Stewart\Contracts\Exception\ExposureException;
11use Stewart\Contracts\Exposure\DeviceInfo;
12use Stewart\Contracts\Exposure\ExposedEntityKey;
13use Stewart\Contracts\Exposure\ExposedEntitySnapshot;
14use Stewart\Contracts\Exposure\ExposedStateChange;
15use Stewart\Contracts\Exposure\ExposureConfig;
16use Stewart\Runtime\Exception\TransportError;
17use Stewart\Runtime\Exception\TransportException;
18use Stewart\Runtime\Ipc\Message\ExposeEntityRequest;
19use Stewart\Runtime\Ipc\Message\ExposeEntityResult;
20use Stewart\Runtime\Ipc\Message\ExposureAcknowledged;
21use Stewart\Runtime\Ipc\Message\ReconfigureExposedEntityRequest;
22use Stewart\Runtime\Ipc\Message\RemoveExposedEntityRequest;
23use Stewart\Runtime\Ipc\Message\UpdateExposedEntityRequest;
24use Stewart\Runtime\Ipc\Message\WorkerMessage;
25use Stewart\Runtime\Ipc\Transport;
26use Stewart\Runtime\Ipc\WorkerSettings;
27use Stewart\Runtime\Model\ResourceScope;
28use Stewart\Runtime\Worker\PendingRequest;
29use Stewart\Runtime\Worker\PendingRequests;
30use Stewart\Support\Time\Deadlines;
31
32final readonly class ExposureRequester
33{
34    public function __construct(
35        private Transport $transport,
36        private PendingRequests $pending,
37        private Deadlines $deadlines,
38        private WorkerSettings $workerSettings,
39    ) {}
40
41    /** @throws ExposureException */
42    public function requestExposure(
43        ResourceScope $scope,
44        ExposedEntityKey $key,
45        ExposureConfig $config,
46        ?DeviceInfo $device,
47        ExposedStateChange $change,
48    ): ?ExposedEntitySnapshot {
49        $pending = $this->pending->open(ExposeEntityResult::class, self::describeUnreachable($key));
50
51        return $this->awaitAnswer($pending, $key, new ExposeEntityRequest($pending->correlationId, $scope, $key, $config, $device, $change))->snapshot;
52    }
53
54    /** @throws ExposureException */
55    public function requestReconfiguration(ResourceScope $scope, ExposedEntityKey $key, ExposureConfig $config): ?ExposedEntitySnapshot
56    {
57        $pending = $this->pending->open(ExposeEntityResult::class, self::describeUnreachable($key));
58
59        return $this->awaitAnswer($pending, $key, new ReconfigureExposedEntityRequest($pending->correlationId, $scope, $key, $config))->snapshot;
60    }
61
62    /** @throws ExposureException */
63    public function requestUpdate(ResourceScope $scope, ExposedEntityKey $key, ExposedStateChange $change): void
64    {
65        $pending = $this->pending->open(ExposureAcknowledged::class, self::describeUnreachable($key));
66        $this->awaitAnswer($pending, $key, new UpdateExposedEntityRequest($pending->correlationId, $scope, $key, $change));
67    }
68
69    /** @throws ExposureException */
70    public function requestRemoval(ResourceScope $scope, ExposedEntityKey $key): void
71    {
72        $pending = $this->pending->open(ExposureAcknowledged::class, self::describeUnreachable($key));
73        $this->awaitAnswer($pending, $key, new RemoveExposedEntityRequest($pending->correlationId, $scope, $key));
74    }
75
76    /** @return Closure(string): ExposureException */
77    private static function describeUnreachable(ExposedEntityKey $key): Closure
78    {
79        return static fn(string $detail): ExposureException => ExposureException::unreachable($key, $detail);
80    }
81
82    /**
83     * @template TResult of object
84     *
85     * @param PendingRequest<TResult> $pending
86     * @return TResult
87     * @throws ExposureException
88     */
89    private function awaitAnswer(PendingRequest $pending, ExposedEntityKey $key, WorkerMessage $request): object
90    {
91        try {
92            $this->transport->send($request);
93
94            return $pending->getFuture()->await($this->deadlines->timeout($this->workerSettings->callTimeout));
95        } catch (TransportException $e) {
96            throw $e->reason === TransportError::Unencodable
97                ? ExposureException::stateInvalid($e->getMessage())
98                : ExposureException::unreachable($key, 'the broker is gone: ' . $e->getMessage(), $e);
99        } catch (CancelledException) {
100            throw ExposureException::createForReason(ExposureError::TimedOut, key: $key->value, timeout: (string) $this->workerSettings->callTimeout);
101        } finally {
102            $this->pending->forget($pending->correlationId);
103        }
104    }
105}