Fix subscribes to multiple topics, and sends multiple heartbeat packe…

…ts (#241). (#244)
streamnative · Nov 1, 2021 · 5afca3f · 5afca3f
1 parent 2829258
commit 5afca3f
Showing 1 changed file with 8 additions and 12 deletions.
diff --git a/...ain/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyProtocolMethodProcessor.java b/...ain/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyProtocolMethodProcessor.java
@@ -175,9 +175,7 @@ public void processPubComp(Channel channel, MqttMessage msg) {
     @Override
     public void processPingReq(Channel channel) {
         channel.writeAndFlush(pingResp());
-        topicBrokers.forEach((k, v) -> v.whenComplete((exchanger, error) -> {
-            exchanger.writeAndFlush(pingReq());
-        }));
+        brokerPool.forEach((k, v) -> v.writeAndFlush(pingReq()));
     }
 
     @Override
@@ -186,12 +184,12 @@ public void processDisconnect(Channel channel, MqttMessage msg) {
         if (log.isDebugEnabled()) {
             log.debug("[Proxy Disconnect] [{}] ", clientId);
         }
-        topicBrokers.forEach((k, v) -> v.whenComplete((exchanger, error) -> {
-            exchanger.writeAndFlush(msg);
-            exchanger.close();
-        }));
-        topicBrokers.clear();
+        brokerPool.forEach((k, v) -> {
+            v.writeAndFlush(msg);
+            v.close();
+        });
         brokerPool.clear();
+        topicBrokers.clear();
         // When login, checkState(msg) failed, connection is null.
         Connection connection = NettyUtils.getConnection(channel);
         if (connection == null) {
@@ -211,11 +209,9 @@ public void processConnectionLost(Channel channel) {
         }
         Connection connection = NettyUtils.getConnection(channel);
         connectionManager.removeConnection(connection);
-        topicBrokers.forEach((k, v) -> v.whenComplete((exchanger, error) -> {
-            exchanger.close();
-        }));
-        topicBrokers.clear();
+        brokerPool.forEach((k, v) -> v.close());
         brokerPool.clear();
+        topicBrokers.clear();
     }
 
     @Override