Lines 87.50% 7 / 8
Methods 80.00% 4 / 5
Classes 0.00% 0 / 1
Name Lines Methods CRAP
 __construct 100.00% 2 / 2 100.00% 1 / 1 1
 pushMessage 100.00% 1 / 1 100.00% 1 / 1 1
 cancelSubscriptions 0.00% 0 / 1 0.00% 0 / 1 2
 publish 100.00% 1 / 1 100.00% 1 / 1 1
 watchMessages 100.00% 3 / 3 100.00% 1 / 1 1
17final class RecordingMqtt implements Mqtt
18{
19    public private(set) MqttMessageCollection $published;
20
21    /** @var PushSource<MqttMessage> */
22    private readonly PushSource $incoming;
23
24    public function __construct(public readonly ManualTimers $timers = new ManualTimers())
25    {
26        $this->published = MqttMessageCollection::empty();
27        $this->incoming = new PushSource($timers->handlerFibers);
28    }
29
30    public function pushMessage(MqttMessage $message): void
31    {
32        $this->incoming->push($message);
33    }
34
35    public function cancelSubscriptions(): void
36    {
37        $this->incoming->cancelAttachments();
38    }
39
40    public function publish(string $topic, string|array $payload, MqttQos $qos = MqttQos::AtMostOnce, bool $retain = false): void
41    {
42        $this->published = $this->published->withMqttMessage(MqttMessage::createForPublish($topic, $payload, $qos, $retain));
43    }
44
45    public function watchMessages(string $topicFilter): EventStream
46    {
47        $filter = Selector::mqttFilter($topicFilter);
48
49        return new OperatorStream($this->incoming, $this->timers)
50            ->filter(static fn(MqttMessage $message): bool => $filter->matches($message->topic));
51    }
52}