Lines 88.37% 152 / 172
Functions and Methods 62.50% 15 / 24
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
HaConnection 88.37% 152 / 172 62.50% 15 / 24 76.49 0.00% 0 / 1
 __construct 100.00% 3 / 3 100.00% 1 / 1 1
 connect 100.00% 26 / 26 100.00% 1 / 1 5
 flushEvents 100.00% 1 / 1 100.00% 1 / 1 1
 send 100.00% 6 / 6 100.00% 1 / 1 2
 subscribeEvents 100.00% 8 / 8 100.00% 1 / 1 2
 unsubscribeEvents 100.00% 2 / 2 100.00% 1 / 1 1
 sendIgnoringResult 100.00% 1 / 1 100.00% 1 / 1 1
 onDisconnect 100.00% 1 / 1 100.00% 1 / 1 1
 isConnected 100.00% 1 / 1 100.00% 1 / 1 1
 close 100.00% 7 / 7 100.00% 1 / 1 1
 openSocket 66.66% 4 / 6 0.00% 0 / 1 3.33
 defaultConnector 100.00% 12 / 12 100.00% 1 / 1 2
 sendAndAwaitResponse 94.44% 17 / 18 0.00% 0 / 1 4.00
 authenticate 78.94% 15 / 19 0.00% 0 / 1 7.46
 receiveMessage 100.00% 2 / 2 100.00% 1 / 1 1
 receiveLoop 100.00% 9 / 9 100.00% 1 / 1 5
 handleConnectionLost 83.33% 10 / 12 0.00% 0 / 1 4.07
 decodeMessageOrLog 25.00% 1 / 4 0.00% 0 / 1 3.69
 dispatchMessage 71.42% 5 / 7 0.00% 0 / 1 5.58
 dispatchEvent 100.00% 6 / 6 100.00% 1 / 1 4
 convertCloseToLoss 100.00% 4 / 4 100.00% 1 / 1 5
 assertSuccess 87.50% 7 / 8 0.00% 0 / 1 6.07
 encodeMessage 33.33% 1 / 3 0.00% 0 / 1 3.19
 decodeMessage 50.00% 3 / 6 0.00% 0 / 1 4.12
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Client\Connection;
6
7use Amp\Cancellation;
8use Amp\CancelledException;
9use Amp\CompositeCancellation;
10use Amp\DeferredCancellation;
11use Amp\Websocket\Client\Rfc6455ConnectionFactory;
12use Amp\Websocket\Client\Rfc6455Connector;
13use Amp\Websocket\Client\WebsocketConnection;
14use Amp\Websocket\Client\WebsocketConnector;
15use Amp\Websocket\Client\WebsocketHandshake;
16use Amp\Websocket\Parser\Rfc6455ParserFactory;
17use Amp\Websocket\PeriodicHeartbeatQueue;
18use Amp\Websocket\WebsocketCloseCode;
19use Amp\Websocket\WebsocketCloseInfo;
20use Closure;
21use JsonException;
22use Psr\Log\LoggerInterface;
23use Psr\Log\NullLogger;
24use Stewart\Client\Connection\Command\Authenticate;
25use Stewart\Client\Connection\Command\HaCommand;
26use Stewart\Client\Connection\Command\SubscriptionCommand;
27use Stewart\Client\Connection\Command\UnsubscribeEvents;
28use Stewart\Client\Exception\HaClientError;
29use Stewart\Client\Exception\HaClientException;
30use Stewart\Support\Json\JsonDecoder;
31use Stewart\Support\Json\JsonEncoder;
32use Stewart\Support\Json\JsonShape;
33use Stewart\Support\Time\Deadlines;
34use Throwable;
35
36use function Amp\async;
37
38/** @internal */
39final class HaConnection
40{
41    private const string UNAUTHORIZED_ERROR_CODE = 'unauthorized';
42
43    private ?WebsocketConnection $socket = null;
44
45    private readonly MessageCorrelator $correlator;
46
47    /** @var array<int, Closure(array<string, mixed>): void> */
48    private array $eventHandlers = [];
49
50    public private(set) ?string $haVersion = null;
51
52    private ?DeferredCancellation $connecting = null;
53
54    /** @var (Closure(HaClientException): void)|null */
55    private ?Closure $onDisconnect = null;
56
57    private EventDelivery $events;
58
59    private ?WebsocketConnector $connector;
60
61    public function __construct(
62        private readonly ConnectionConfig $config,
63        private readonly Deadlines $deadlines,
64        private readonly LoggerInterface $logger = new NullLogger(),
65        ?WebsocketConnector $connector = null,
66    ) {
67        $this->connector = $connector;
68        $this->correlator = new MessageCorrelator();
69        $this->events = new EventDelivery($config->eventBufferLimit, $logger);
70    }
71
72    public function connect(): void
73    {
74        if ($this->socket !== null) {
75            $this->close();
76        }
77
78        $this->logger->debug('Connecting to Home Assistant', ['url' => (string) $this->config->url]);
79
80        $this->connecting = $connecting = new DeferredCancellation();
81        $cancellation = new CompositeCancellation(
82            $connecting->getCancellation(),
83            $this->deadlines->timeout($this->config->connectTimeout),
84        );
85
86        try {
87            $socket = $this->openSocket($cancellation);
88
89            try {
90                $this->authenticate($socket, $cancellation);
91            } catch (Throwable $e) {
92                $socket->close();
93
94                throw $e;
95            }
96        } catch (CancelledException $e) {
97            throw $connecting->isCancelled()
98                ? HaClientException::createForReason(HaClientError::ClosedLocally, phase: 'it was being established')
99                : HaClientException::createForReason(
100                    HaClientError::ConnectTimedOut,
101                    $e,
102                    url: (string) $this->config->url,
103                    timeout: (string) $this->config->connectTimeout,
104                );
105        } finally {
106            $this->connecting = null;
107        }
108
109        $this->socket = $socket;
110        $this->events = new EventDelivery($this->config->eventBufferLimit, $this->logger);
111
112        async(fn() => $this->receiveLoop($socket))->ignore();
113    }
114
115    public function flushEvents(): void
116    {
117        $this->events->flush();
118    }
119
120    /** @return array<array-key, mixed> */
121    public function send(HaCommand $command): array
122    {
123        $response = $this->sendAndAwaitResponse($this->correlator->nextId(), $command);
124        self::assertSuccess($response, $command);
125
126        $result = $response['result'] ?? null;
127
128        return \is_array($result)
129            ? $result
130            : throw HaClientException::protocolViolation(\sprintf('a non-object result for %s', $command->describe()));
131    }
132
133    /** @param Closure(array<string, mixed>): void $handler */
134    public function subscribeEvents(SubscriptionCommand $command, Closure $handler): int
135    {
136        $id = $this->correlator->nextId();
137        $this->eventHandlers[$id] = $handler;
138
139        try {
140            self::assertSuccess($this->sendAndAwaitResponse($id, $command), $command);
141        } catch (Throwable $e) {
142            unset($this->eventHandlers[$id]);
143
144            throw $e;
145        }
146
147        $this->logger->debug('Subscribed to Home Assistant events', ['command' => $command->describe(), 'id' => $id]);
148
149        return $id;
150    }
151
152    public function unsubscribeEvents(int $subscriptionId): void
153    {
154        unset($this->eventHandlers[$subscriptionId]);
155
156        $this->sendIgnoringResult(new UnsubscribeEvents($subscriptionId));
157    }
158
159    public function sendIgnoringResult(HaCommand $command): void
160    {
161        self::assertSuccess($this->sendAndAwaitResponse($this->correlator->nextId(), $command), $command);
162    }
163
164    /** @param Closure(HaClientException): void $handler */
165    public function onDisconnect(Closure $handler): void
166    {
167        $this->onDisconnect = $handler;
168    }
169
170    public function isConnected(): bool
171    {
172        return $this->socket !== null;
173    }
174
175    public function close(): void
176    {
177        $this->connecting?->cancel();
178
179        $socket = $this->socket;
180        $this->socket = null;
181        $this->eventHandlers = [];
182
183        $this->events->close();
184        $this->correlator->failAll(HaClientException::createForReason(HaClientError::ClosedLocally, phase: 'commands were waiting for an answer'));
185
186        $socket?->close();
187    }
188
189    private function openSocket(Cancellation $cancellation): WebsocketConnection
190    {
191        $this->connector ??= $this->defaultConnector();
192
193        try {
194            return $this->connector->connect(new WebsocketHandshake($this->config->url->reveal()), $cancellation);
195        } catch (CancelledException $e) {
196            throw $e;
197        } catch (Throwable $e) {
198            throw HaClientException::createForReason(HaClientError::ConnectFailed, $e, url: (string) $this->config->url);
199        }
200    }
201
202    private function defaultConnector(): WebsocketConnector
203    {
204        $interval = $this->config->heartbeatInterval;
205
206        return new Rfc6455Connector(
207            connectionFactory: new Rfc6455ConnectionFactory(
208                heartbeatQueue: $interval === null
209                    ? null
210                    : new PeriodicHeartbeatQueue(max(1, $this->config->heartbeatMissedLimit), max(1, (int) round($interval->toSeconds()))),
211                parserFactory: new Rfc6455ParserFactory(
212                    messageSizeLimit: $this->config->messageSizeLimit,
213                    frameSizeLimit: $this->config->frameSizeLimit,
214                ),
215            ),
216        );
217    }
218
219    /** @return array<string, mixed> */
220    private function sendAndAwaitResponse(int $id, HaCommand $command): array
221    {
222        $socket = $this->socket ?? throw HaClientException::createForReason(HaClientError::NotConnected);
223        $description = $command->describe();
224        $payload = self::encodeMessage([...$command->toMessage(), 'id' => $id], $description);
225        $future = $this->correlator->expect($id);
226
227        try {
228            try {
229                $socket->sendText($payload);
230            } catch (Throwable $e) {
231                // A close or drop during the send already failed this command with the real cause.
232                if ($future->isComplete()) {
233                    $future->await();
234                }
235
236                throw HaClientException::createForReason(HaClientError::SendFailed, $e, command: $description);
237            }
238
239            return $future->await($this->deadlines->timeout($this->config->commandTimeout));
240        } catch (CancelledException $e) {
241            throw HaClientException::createForReason(
242                HaClientError::CommandTimedOut,
243                $e,
244                command: $description,
245                timeout: (string) $this->config->commandTimeout,
246            );
247        } finally {
248            $this->correlator->forget($id);
249        }
250    }
251
252    private function authenticate(WebsocketConnection $socket, Cancellation $cancellation): void
253    {
254        $greeting = $this->receiveMessage($socket, $cancellation);
255
256        if (($greeting['type'] ?? null) !== 'auth_required') {
257            throw HaClientException::createForReason(
258                HaClientError::UnexpectedGreeting,
259                receivedType: \is_scalar($greeting['type'] ?? null) ? (string) $greeting['type'] : 'nothing',
260            );
261        }
262
263        $this->haVersion = \is_string($greeting['ha_version'] ?? null) ? $greeting['ha_version'] : 'unknown';
264
265        $socket->sendText(self::encodeMessage(new Authenticate($this->config->token)->toMessage(), 'auth'));
266
267        $response = $this->receiveMessage($socket, $cancellation);
268
269        if (($response['type'] ?? null) !== 'auth_ok') {
270            throw HaClientException::createForReason(
271                HaClientError::TokenRejected,
272                reason: \is_string($response['message'] ?? null) ? $response['message'] : 'no reason given',
273            );
274        }
275
276        $this->haVersion = \is_string($response['ha_version'] ?? null) ? $response['ha_version'] : $this->haVersion;
277
278        $this->logger->info('Authenticated with Home Assistant', [
279            'url' => (string) $this->config->url,
280            'ha_version' => $this->haVersion,
281        ]);
282    }
283
284    /** @return array<string, mixed> */
285    private function receiveMessage(WebsocketConnection $socket, Cancellation $cancellation): array
286    {
287        $message = $socket->receive($cancellation) ?? throw HaClientException::createForReason(HaClientError::ClosedDuringAuthentication);
288
289        return self::decodeMessage($message->buffer());
290    }
291
292    private function receiveLoop(WebsocketConnection $socket): void
293    {
294        try {
295            while (($message = $socket->receive()) !== null) {
296                $decoded = $this->decodeMessageOrLog($message->buffer());
297
298                if ($decoded !== null) {
299                    $this->dispatchMessage($decoded);
300                }
301            }
302
303            $lost = self::convertCloseToLoss($socket->getCloseInfo());
304        } catch (Throwable $e) {
305            $lost = $e instanceof HaClientException ? $e : HaClientException::connectionDropped('socket failed: ' . $e->getMessage(), $e);
306        }
307
308        $this->handleConnectionLost($socket, $lost);
309        $socket->close();
310    }
311
312    private function handleConnectionLost(WebsocketConnection $socket, HaClientException $lost): void
313    {
314        if ($this->socket !== $socket) {
315            return;
316        }
317
318        $this->socket = null;
319        $this->eventHandlers = [];
320        $this->events->close();
321        $abandoned = $this->correlator->failAll($lost);
322
323        $this->logger->warning($lost->getMessage(), ['abandoned_commands' => $abandoned]);
324
325        if ($this->onDisconnect === null) {
326            return;
327        }
328
329        try {
330            ($this->onDisconnect)($lost);
331        } catch (Throwable $e) {
332            $this->logger->error('Disconnect handler failed', ['exception' => $e]);
333        }
334    }
335
336    /** @return array<string, mixed>|null */
337    private function decodeMessageOrLog(string $payload): ?array
338    {
339        try {
340            return self::decodeMessage($payload);
341        } catch (HaClientException $e) {
342            $this->logger->error('Failed handling a Home Assistant message', ['exception' => $e]);
343
344            return null;
345        }
346    }
347
348    /** @param array<string, mixed> $message */
349    private function dispatchMessage(array $message): void
350    {
351        $id = $message['id'] ?? null;
352
353        if (!\is_int($id)) {
354            return;
355        }
356
357        match ($message['type'] ?? null) {
358            'result' => $this->correlator->resolve($id, $message),
359            'event' => $this->dispatchEvent($id, $message),
360            default => null,
361        };
362    }
363
364    /** @param array<string, mixed> $message */
365    private function dispatchEvent(int $id, array $message): void
366    {
367        $handler = $this->eventHandlers[$id] ?? null;
368        $event = $message['event'] ?? null;
369
370        if ($handler === null || !\is_array($event)) {
371            return;
372        }
373
374        if ($this->events->enqueue($handler, JsonShape::treatKeysAsStrings($event)) === EnqueueOutcome::Full) {
375            throw HaClientException::createForReason(HaClientError::EventBacklogExceeded, limit: $this->config->eventBufferLimit);
376        }
377    }
378
379    // amphp reports its own parser's size check as a close by the peer, so the code decides that case.
380    private static function convertCloseToLoss(WebsocketCloseInfo $close): HaClientException
381    {
382        $reason = $close->getReason();
383
384        return match (true) {
385            $close->getCode() === WebsocketCloseCode::MESSAGE_TOO_LARGE => HaClientException::createForReason(HaClientError::MessageTooLarge),
386            $close->isByPeer() => HaClientException::connectionDropped($reason === '' ? 'closed by Home Assistant' : 'closed by Home Assistant: ' . $reason),
387            default => HaClientException::connectionDropped(ltrim(\sprintf('%s (close code %d)', $reason, $close->getCode()))),
388        };
389    }
390
391    /** @param array<string, mixed> $response */
392    private static function assertSuccess(array $response, HaCommand $command): void
393    {
394        if (($response['success'] ?? false) === true) {
395            return;
396        }
397
398        $error = \is_array($response['error'] ?? null) ? $response['error'] : [];
399        $message = \is_string($error['message'] ?? null) ? $error['message'] : 'unknown error';
400        $code = \is_scalar($error['code'] ?? null) ? (string) $error['code'] : null;
401
402        throw $code === self::UNAUTHORIZED_ERROR_CODE
403            ? HaClientException::createForReason(HaClientError::CommandUnauthorized, command: $command->describe(), detail: $message, errorCode: $code)
404            : HaClientException::createForReason(HaClientError::CommandRejected, command: $command->describe(), detail: $message, errorCode: $code);
405    }
406
407    /** @param array<string, mixed> $message */
408    private static function encodeMessage(array $message, string $description): string
409    {
410        try {
411            return JsonEncoder::encodeToJson($message);
412        } catch (JsonException $e) {
413            throw HaClientException::commandUnencodable($description, $e);
414        }
415    }
416
417    /** @return array<string, mixed> */
418    private static function decodeMessage(string $payload): array
419    {
420        try {
421            $decoded = JsonDecoder::decodeJson($payload);
422        } catch (JsonException $e) {
423            throw HaClientException::protocolViolation('a message that is not JSON', $e);
424        }
425
426        if (!\is_array($decoded)) {
427            throw HaClientException::protocolViolation('a non-object message');
428        }
429
430        return JsonShape::treatKeysAsStrings($decoded);
431    }
432}