Lines 100.00% 17 / 17
Functions and Methods 100.00% 8 / 8
Classes and Traits 100.00% 1 / 1
Name Lines Functions and Methods CRAP Classes and Traits
OutboundMqttQueue 100.00% 17 / 17 100.00% 8 / 8 11 100.00% 1 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 canHoldMessages 100.00% 1 / 1 100.00% 1 / 1 1
 isFull 100.00% 1 / 1 100.00% 1 / 1 1
 enqueueMessage 100.00% 5 / 5 100.00% 1 / 1 2
 recordSendAttempt 100.00% 1 / 1 100.00% 1 / 1 1
 acknowledgeMessage 100.00% 1 / 1 100.00% 1 / 1 1
 replayPendingMessages 100.00% 6 / 6 100.00% 1 / 1 3
 countPending 100.00% 1 / 1 100.00% 1 / 1 1
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Mqtt;
6
7use Closure;
8use Stewart\Contracts\Mqtt\MqttMessage;
9
10// QoS 1 messages stay here until the server acknowledges them, so a reconnect can send them again.
11final class OutboundMqttQueue
12{
13    // A message the server keeps dropping the connection over would otherwise be replayed forever.
14    private const int MAX_SEND_ATTEMPTS = 5;
15
16    /** @var array<int, PendingMqttMessage> */
17    private array $pendingBySequence = [];
18
19    private int $nextSequence = 0;
20
21    public function __construct(private readonly int $outboundCapacity) {}
22
23    public function canHoldMessages(): bool
24    {
25        return $this->outboundCapacity > 0;
26    }
27
28    public function isFull(): bool
29    {
30        return \count($this->pendingBySequence) >= $this->outboundCapacity;
31    }
32
33    public function enqueueMessage(MqttMessage $message): int
34    {
35        $sequence = $this->nextSequence++;
36        $this->pendingBySequence[$sequence] = new PendingMqttMessage($message);
37
38        if (\count($this->pendingBySequence) > $this->outboundCapacity) {
39            $this->pendingBySequence = \array_slice($this->pendingBySequence, -$this->outboundCapacity, preserve_keys: true);
40        }
41
42        return $sequence;
43    }
44
45    public function recordSendAttempt(int $sequence): void
46    {
47        ($this->pendingBySequence[$sequence] ?? null)?->recordSendAttempt();
48    }
49
50    public function acknowledgeMessage(int $sequence): void
51    {
52        unset($this->pendingBySequence[$sequence]);
53    }
54
55    /**
56     * @param Closure(int, MqttMessage): void $send
57     * @param Closure(MqttMessage, int): void $onGivenUp
58     */
59    public function replayPendingMessages(Closure $send, Closure $onGivenUp): void
60    {
61        foreach ($this->pendingBySequence as $sequence => $pending) {
62            if ($pending->sendAttempts >= self::MAX_SEND_ATTEMPTS) {
63                unset($this->pendingBySequence[$sequence]);
64                $onGivenUp($pending->message, $pending->sendAttempts);
65
66                continue;
67            }
68
69            $send($sequence, $pending->message);
70        }
71    }
72
73    public function countPending(): int
74    {
75        return \count($this->pendingBySequence);
76    }
77}