Fix connection management

Author: coderzc

mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/MQTTConnectionManager.java

@@ -23,6 +23,9 @@
 import io.streamnative.pulsar.handlers.mqtt.common.systemtopic.MqttEvent;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
@@ -36,6 +39,8 @@
 @Slf4j
 public class MQTTConnectionManager {
 
+    private static final Connection REMOTE_CONNECTION_SOLT = Connection.builder().build();
+
     private final ConcurrentMap<String, Connection> localConnections;
 
     private final ConcurrentMap<String, Connection> eventConnections;
@@ -102,11 +107,11 @@ public Collection<Connection> getLocalConnections() {
         return this.localConnections.values();
     }
 
-    public Collection<Connection> getAllConnections() {
-        Collection<Connection> connections = new ArrayList<>(this.localConnections.values().size()
-                    + this.eventConnections.values().size());
-        connections.addAll(this.localConnections.values());
-        connections.addAll(eventConnections.values());
+    public Collection<String> getAllConnectionsId() {
+        Set<String> connections = new LinkedHashSet<>(this.localConnections.keySet().size()
+                    + this.eventConnections.keySet().size());
+        connections.addAll(this.localConnections.keySet());
+        connections.addAll(eventConnections.keySet());
         return connections;
     }
 
@@ -126,7 +131,7 @@ public void onChange(MqttEvent event) {
                         log.warn("[ConnectEvent] close existing connection : {}", connection);
                         connection.disconnect();
                     } else {
-                        eventConnections.put(connectEvent.getClientId(), connection);
+                        eventConnections.put(connectEvent.getClientId(), REMOTE_CONNECTION_SOLT);
                     }
                 }
             }

mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/systemtopic/SystemTopicBasedSystemEventService.java

@@ -237,7 +237,13 @@ private void refreshCache(Message<MqttEvent> msg) {
                 default:
                     break;
             }
-            listeners.forEach(listener -> listener.onChange(value));
+            listeners.forEach(listener -> {
+                try {
+                    listener.onChange(value);
+                } catch (Throwable e) {
+                    log.error("Failed to process event : {}", value.getKey(), e);
+                }
+            });
         } catch (Throwable ex) {
             log.error("refresh cache error", ex);
         }

mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/web/admin/Devices.java

@@ -45,9 +45,8 @@ public class Devices extends WebResource {
             @ApiResponse(code = 500, message = "Internal server error")})
     public void getList(@Suspended final AsyncResponse asyncResponse) {
         try {
-            final Collection<Connection> allConnections = service().getConnectionManager().getAllConnections();
-            asyncResponse.resume(allConnections.stream().map(e ->
-                    e.getClientId()).collect(Collectors.toList()));
+            final Collection<String> allConnections = service().getConnectionManager().getAllConnectionsId();
+            asyncResponse.resume(allConnections);
         } catch (Exception e) {
             log.error("[{}] Failed to list devices {}", clientAppId(), e);
             asyncResponse.resume(new RestException(e));