Update getMqttClient and Add getSubTopic, getSubQos, getUnsubscribe

Author: sy-records

src/Handler/AbstractHandler.php

@@ -97,6 +97,29 @@ public function getTopic(): string
         return (string) $this->input->getOption('topic');
     }
 
+    public function getSubTopic(): array
+    {
+        return $this->input->getOption('topic');
+    }
+
+    public function getSubQos(): array
+    {
+        $qos = $this->input->getOption('qos');
+
+        $subQos = [];
+        if ($this->getProtocolLevel() === ProtocolInterface::MQTT_PROTOCOL_LEVEL_5_0) {
+            foreach ($qos as $item) {
+                $subQos[] = ['qos' => (int) $item];
+            }
+        } else {
+            foreach ($qos as $item) {
+                $subQos[] = (int) $item;
+            }
+        }
+
+        return $subQos;
+    }
+
     public function getMessage(): string
     {
         return (string) $this->input->getOption('message');
@@ -153,9 +176,9 @@ public function getConnectConfig(): ClientConfig
             ->setMaxAttempts(0); // Disable auto reconnection
     }
 
-    public function getMqttClient(): Client
+    public function getMqttClient(string $host, int $port, ClientConfig $config): Client
     {
-        return new Client($this->getHost(), $this->getPort(), $this->getConnectConfig());
+        return new Client($host, $port, $config);
     }
 
     public function genWillData(): array
@@ -201,6 +224,11 @@ public function getProperties(string $key = ''): array
         return [];
     }
 
+    public function getUnsubscribe(): array
+    {
+        return $this->input->getOption('unsubscribe');
+    }
+
     protected function log($msg): void
     {
         $date = date('Y-m-d H:i:s');

src/Handler/PublishHandler.php

@@ -34,11 +34,12 @@ public function handle(InputInterface $input, OutputInterface $output): int
         }
 
         try {
-            $client = $this->getMqttClient();
+            $config = $this->getConnectConfig();
+            $client = $this->getMqttClient($this->getHost(), $this->getPort(), $config);
             $connect = $client->connect($this->getCleanSession(), $this->genWillData());
             $this->logInfo("Connect {$this->getHost()} successfully, recv: ");
             $this->log(json_encode($connect));
-            if ($this->getConnectConfig()->isMQTT5()) {
+            if ($config->isMQTT5()) {
                 $this->logInfo("Connect Reason Code: {$connect['code']}, Reason: " . ReasonCode::getReasonPhrase($connect['code']));
             }
             $publish = $client->publish($topic, $message, $this->getQos(), $this->getDup(), $this->getRetain(), $this->getProperties('publish'));
@@ -50,7 +51,7 @@ public function handle(InputInterface $input, OutputInterface $output): int
         if (is_array($publish)) {
             $this->logInfo("Publish message '{$message}' to '{$topic}', recv: ");
             $this->log(json_encode($publish));
-            if ($this->getConnectConfig()->isMQTT5()) {
+            if ($config->isMQTT5()) {
                 $this->logInfo("Publish Reason Code: {$publish['code']}, Reason: " . ReasonCode::getReasonPhrase($publish['code']));
                 if ($publish['code']) {
                     goto failure;