Lines 94.44% 34 / 36
Functions and Methods 71.42% 5 / 7
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
EventDelivery 94.44% 34 / 36 71.42% 5 / 7 19.06 0.00% 0 / 1
 __construct 100.00% 2 / 2 100.00% 1 / 1 1
 enqueue 100.00% 6 / 6 100.00% 1 / 1 3
 flush 100.00% 7 / 7 100.00% 1 / 1 5
 close 80.00% 4 / 5 0.00% 0 / 1 2.03
 enqueueDelivery 100.00% 5 / 5 100.00% 1 / 1 2
 deliverQueuedEvents 90.00% 9 / 10 0.00% 0 / 1 5.03
 createDroppedException 100.00% 1 / 1 100.00% 1 / 1 1
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Client\Connection;
6
7use Amp\Future;
8use Amp\Pipeline\ConcurrentIterator;
9use Amp\Pipeline\DisposedException;
10use Amp\Pipeline\Queue;
11use Closure;
12use Fiber;
13use Psr\Log\LoggerInterface;
14use Stewart\Client\Exception\HaClientError;
15use Stewart\Client\Exception\HaClientException;
16use Throwable;
17
18use function Amp\async;
19
20/** @internal */
21final class EventDelivery
22{
23    /** @var Queue<Closure(): void> */
24    private readonly Queue $queue;
25
26    /** @var ConcurrentIterator<Closure(): void> */
27    private readonly ConcurrentIterator $deliveries;
28
29    /** @var Fiber<mixed, mixed, mixed, mixed>|null */
30    private ?Fiber $deliverer = null;
31
32    private bool $started = false;
33
34    private bool $closed = false;
35
36    private int $queued = 0;
37
38    public function __construct(
39        private readonly int $limit,
40        private readonly LoggerInterface $logger,
41    ) {
42        $this->queue = new Queue();
43        $this->deliveries = $this->queue->iterate();
44    }
45
46    /**
47     * @param Closure(array<string, mixed>): void $handler
48     * @param array<string, mixed> $payload
49     */
50    public function enqueue(Closure $handler, array $payload): EnqueueOutcome
51    {
52        if ($this->closed) {
53            return EnqueueOutcome::Closed;
54        }
55
56        if ($this->queued >= $this->limit) {
57            return EnqueueOutcome::Full;
58        }
59
60        $this->enqueueDelivery(static fn() => $handler($payload))->ignore();
61
62        return EnqueueOutcome::Accepted;
63    }
64
65    /** @throws HaClientException */
66    public function flush(): void
67    {
68        if ($this->deliverer !== null && $this->deliverer === Fiber::getCurrent()) {
69            throw HaClientException::createForReason(HaClientError::EventQueueReentered);
70        }
71
72        if ($this->closed) {
73            throw self::createDroppedException();
74        }
75
76        try {
77            // Resolves when the consumer takes this marker, which is after every earlier event.
78            $this->enqueueDelivery(static function (): void {})->await();
79        } catch (DisposedException) {
80            throw self::createDroppedException();
81        }
82    }
83
84    public function close(): void
85    {
86        if ($this->closed) {
87            return;
88        }
89
90        $this->closed = true;
91        $this->queued = 0;
92        $this->deliveries->dispose();
93    }
94
95    /**
96     * @param Closure(): void $delivery
97     * @return Future<null>
98     */
99    private function enqueueDelivery(Closure $delivery): Future
100    {
101        ++$this->queued;
102
103        if (!$this->started) {
104            $this->started = true;
105            async($this->deliverQueuedEvents(...))->ignore();
106        }
107
108        return $this->queue->pushAsync($delivery);
109    }
110
111    private function deliverQueuedEvents(): void
112    {
113        $this->deliverer = Fiber::getCurrent();
114
115        try {
116            while ($this->deliveries->continue()) {
117                // continue() still yields queued values after disposal.
118                if ($this->closed) {
119                    return;
120                }
121
122                --$this->queued;
123
124                try {
125                    ($this->deliveries->getValue())();
126                } catch (Throwable $e) {
127                    $this->logger->error('A Home Assistant event handler failed', ['exception' => $e]);
128                }
129            }
130        } catch (DisposedException) {
131        } finally {
132            $this->deliverer = null;
133        }
134    }
135
136    private static function createDroppedException(): HaClientException
137    {
138        return HaClientException::connectionDropped('closed while events were being flushed');
139    }
140}