Lines 89.28% 25 / 28
Functions and Methods 60.00% 3 / 5
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
MqttConnector 89.28% 25 / 28 60.00% 3 / 5 9.10 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 openConnection 100.00% 10 / 10 100.00% 1 / 1 2
 acceptHandshake 83.33% 5 / 6 0.00% 0 / 1 3.04
 encodeConnect 100.00% 7 / 7 100.00% 1 / 1 1
 openSocket 50.00% 2 / 4 0.00% 0 / 1 2.50
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Mqtt\Connection;
6
7use Amp\ByteStream\BufferedReader;
8use Amp\Cancellation;
9use Amp\CompositeCancellation;
10use Amp\Socket\ClientTlsContext;
11use Amp\Socket\ConnectContext;
12use Amp\Socket\Socket;
13use Closure;
14use Stewart\Contracts\Mqtt\MqttMessage;
15use Stewart\Contracts\Time\Timers;
16use Stewart\Mqtt\Exception\MqttClientError;
17use Stewart\Mqtt\Exception\MqttClientException;
18use Stewart\Mqtt\Packet\ConnAckPacket;
19use Stewart\Mqtt\Packet\ConnectReturnCode;
20use Stewart\Mqtt\Packet\PacketEncoder;
21use Stewart\Mqtt\Packet\PacketReader;
22use Stewart\Runtime\Config\MqttConfig;
23use Stewart\Support\Time\Deadlines;
24use Throwable;
25
26use function Amp\Socket\connect;
27use function Amp\Socket\connectTls;
28
29final readonly class MqttConnector
30{
31    public function __construct(
32        private PacketReader $packetReader,
33        private PacketEncoder $encoder,
34        private Timers $timers,
35        private Deadlines $deadlines,
36    ) {}
37
38    /**
39     * @param Closure(MqttMessage): void $onMessage
40     * @throws MqttClientException|Throwable
41     */
42    public function openConnection(MqttConfig $config, Closure $onMessage, Cancellation $stop): MqttConnection
43    {
44        $timeout = $this->deadlines->timeout($config->connectTimeout);
45        $socket = $this->openSocket($config, new CompositeCancellation($stop, $timeout));
46        $reader = new BufferedReader($socket);
47
48        try {
49            // Only the timeout ends the handshake: abandoning it after CONNECT would make the server publish the will.
50            $this->acceptHandshake($socket, $reader, $this->encodeConnect($config), $timeout);
51        } catch (Throwable $e) {
52            $socket->close();
53
54            throw $e;
55        }
56
57        $connection = new MqttConnection($socket, $reader, $this->packetReader, $this->encoder, $this->timers, $config->keepalive, $onMessage);
58        $connection->startReading();
59
60        return $connection;
61    }
62
63    /** @throws MqttClientException|Throwable */
64    private function acceptHandshake(Socket $socket, BufferedReader $reader, string $connect, Cancellation $timeout): void
65    {
66        $socket->write($connect);
67        $answer = $this->packetReader->readPacket($reader, $timeout);
68
69        if (!$answer instanceof ConnAckPacket) {
70            throw MqttClientException::packetUnexpected($answer->getPacketType()->name);
71        }
72
73        if ($answer->returnCode !== ConnectReturnCode::Accepted) {
74            throw MqttClientException::createForReason(MqttClientError::ConnectionRefused, refusal: $answer->returnCode->describeRefusal());
75        }
76    }
77
78    private function encodeConnect(MqttConfig $config): string
79    {
80        return $this->encoder->encodeConnect(
81            clientId: $config->clientId,
82            keepalive: $config->keepalive,
83            username: $config->serverUrl->username,
84            password: $config->serverUrl->revealPassword(),
85            will: $config->will,
86        );
87    }
88
89    /** @throws Throwable */
90    private function openSocket(MqttConfig $config, Cancellation $deadline): Socket
91    {
92        if (!$config->serverUrl->usesTls) {
93            return connect($config->serverUrl->toSocketUri(), null, $deadline);
94        }
95
96        $tls = new ConnectContext()->withTlsContext(new ClientTlsContext($config->serverUrl->host));
97
98        return connectTls($config->serverUrl->toSocketUri(), $tls, $deadline);
99    }
100}