fix(网络组件): 修复关闭mqtt网关禁用逻辑错误 (#512)

jetlinks · May 24, 2024 · 53a5d0c · 53a5d0c
1 parent 2b8f571
commit 53a5d0c
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/...t-component/src/main/java/org/jetlinks/community/network/mqtt/client/VertxMqttClient.java b/...t-component/src/main/java/org/jetlinks/community/network/mqtt/client/VertxMqttClient.java
@@ -116,7 +116,7 @@ public void setClient(io.vertx.mqtt.MqttClient client) {
     private void reSubscribe() {
         subscriber
             .getAllSubscriber()
-            .filter(topic -> topic.getSubscribers().size() > 0)
+            .filter(topic -> !topic.getSubscribers().isEmpty())
             .collectMap(topic -> getCompleteTopic(convertMqttTopic(topic.getSubscribers().iterator().next().getT1())),
                         topic -> topic.getSubscribers().iterator().next().getT3())
             .filter(MapUtils::isNotEmpty)
@@ -170,10 +170,10 @@ public Flux<MqttMessage> subscribe(List<String> topics, int qos) {
 
                 Tuple3<String, FluxSink<MqttMessage>, Integer> topicQos = Tuples.of(topic, sink, qos);
 
-                boolean first = sinkTopic.getSubscribers().size() == 0;
+                boolean first = sinkTopic.getSubscribers().isEmpty();
                 sinkTopic.subscribe(topicQos);
                 composite.add(() -> {
-                    if (sinkTopic.unsubscribe(topicQos).size() > 0 && isAlive()) {
+                    if (!sinkTopic.unsubscribe(topicQos).isEmpty() && isAlive() && sinkTopic.getSubscribers().isEmpty()) {
                         client.unsubscribe(convertMqttTopic(completeTopic), result -> {
                             if (result.succeeded()) {
                                 log.debug("unsubscribe mqtt topic {}", completeTopic);