Add mop proxy admin part (#1575)

streamnative · Dec 30, 2024 · d4f96c3 · d4f96c3
1 parent 6407272
commit d4f96c3
Show file tree

Hide file tree

Showing 13 changed files with 672 additions and 14 deletions.
diff --git a/...on/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/MQTTCommonConfiguration.java b/...on/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/MQTTCommonConfiguration.java
@@ -17,6 +17,7 @@
 import com.google.common.collect.Sets;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import lombok.Getter;
 import lombok.Setter;
@@ -376,6 +377,12 @@ public class MQTTCommonConfiguration extends ServiceConfiguration {
     )
     private boolean systemEventEnabled = true;
 
+    @FieldContext(
+            category = CATEGORY_MQTT_PROXY,
+            doc = "Enable web http service."
+    )
+    private Optional<Integer> mopWebServicePort = Optional.of(5680);
+
     public long getMqttTlsCertRefreshCheckDurationSec() {
         if (mqttTlsCertRefreshCheckDurationSec != 300) {
             return mqttTlsCertRefreshCheckDurationSec;

diff --git a/...mmon/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/MQTTConnectionManager.java b/...mmon/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/MQTTConnectionManager.java
@@ -14,12 +14,15 @@
 package io.streamnative.pulsar.handlers.mqtt.common;
 
 import static io.streamnative.pulsar.handlers.mqtt.common.systemtopic.EventType.CONNECT;
+import static io.streamnative.pulsar.handlers.mqtt.common.systemtopic.EventType.DISCONNECT;
 import io.netty.util.HashedWheelTimer;
 import io.netty.util.Timeout;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import io.streamnative.pulsar.handlers.mqtt.common.systemtopic.ConnectEvent;
 import io.streamnative.pulsar.handlers.mqtt.common.systemtopic.EventListener;
 import io.streamnative.pulsar.handlers.mqtt.common.systemtopic.MqttEvent;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
@@ -33,26 +36,33 @@
 @Slf4j
 public class MQTTConnectionManager {
 
-    private final ConcurrentMap<String, Connection> connections;
+    private final ConcurrentMap<String, Connection> localConnections;
+
+    private final ConcurrentMap<String, Connection> eventConnections;
 
     @Getter
     private static final HashedWheelTimer sessionExpireInterval =
             new HashedWheelTimer(
                     new DefaultThreadFactory("session-expire-interval"), 1, TimeUnit.SECONDS);
 
     @Getter
-    private final EventListener eventListener;
+    private final EventListener connectListener;
+
+    @Getter
+    private final EventListener disconnectListener;
 
     private final String advertisedAddress;
 
     public MQTTConnectionManager(String advertisedAddress) {
         this.advertisedAddress = advertisedAddress;
-        this.connections = new ConcurrentHashMap<>(2048);
-        this.eventListener = new ConnectEventListener();
+        this.localConnections = new ConcurrentHashMap<>(2048);
+        this.eventConnections = new ConcurrentHashMap<>(2048);
+        this.connectListener = new ConnectEventListener();
+        this.disconnectListener = new DisconnectEventListener();
     }
 
     public void addConnection(Connection connection) {
-        Connection existing = connections.put(connection.getClientId(), connection);
+        Connection existing = localConnections.put(connection.getClientId(), connection);
         if (existing != null) {
             log.warn("The clientId is existed. Close existing connection. CId={}", existing.getClientId());
             existing.disconnect();
@@ -68,7 +78,7 @@ public void addConnection(Connection connection) {
      */
     public void newSessionExpireInterval(Consumer<Timeout> task, String clientId, int interval) {
         sessionExpireInterval.newTimeout(timeout -> {
-            Connection connection = connections.get(clientId);
+            Connection connection = localConnections.get(clientId);
             if (connection != null
                     && connection.getState() != Connection.ConnectionState.DISCONNECTED) {
                 return;
@@ -80,16 +90,28 @@ public void newSessionExpireInterval(Consumer<Timeout> task, String clientId, in
     // Must use connections.remove(key, value).
     public void removeConnection(Connection connection) {
         if (connection != null) {
-            connections.remove(connection.getClientId(), connection);
+            localConnections.remove(connection.getClientId(), connection);
         }
     }
 
     public Connection getConnection(String clientId) {
-        return connections.get(clientId);
+        return localConnections.get(clientId);
+    }
+
+    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());
+        return connections;
     }
 
     public void close() {
-        connections.values().forEach(connection -> connection.getChannel().close());
+        localConnections.values().forEach(connection -> connection.getChannel().close());
     }
 
     class ConnectEventListener implements EventListener {
@@ -103,9 +125,25 @@ public void onChange(MqttEvent event) {
                     if (connection != null) {
                         log.warn("[ConnectEvent] close existing connection : {}", connection);
                         connection.disconnect();
+                    } else {
+                        eventConnections.put(connectEvent.getClientId(), connection);
                     }
                 }
             }
         }
     }
+
+    //TODO
+    class DisconnectEventListener implements EventListener {
+
+        @Override
+        public void onChange(MqttEvent event) {
+            if (event.getEventType() == DISCONNECT) {
+                ConnectEvent connectEvent = (ConnectEvent) event.getSourceEvent();
+                if (!connectEvent.getAddress().equals(advertisedAddress)) {
+                    eventConnections.remove(connectEvent.getClientId());
+                }
+            }
+        }
+    }
 }
diff --git a/...mmon/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/systemtopic/EventType.java b/...mmon/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/systemtopic/EventType.java
@@ -15,6 +15,7 @@
 
 public enum EventType {
     CONNECT,
+    DISCONNECT,
     LAST_WILL_MESSAGE,
     RETAINED_MESSAGE,
     ADD_PSK_IDENTITY;

diff --git a/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyService.java b/mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/MQTTProxyService.java
@@ -35,6 +35,7 @@
 import io.streamnative.pulsar.handlers.mqtt.proxy.handler.LookupHandler;
 import io.streamnative.pulsar.handlers.mqtt.proxy.handler.PulsarServiceLookupHandler;
 import io.streamnative.pulsar.handlers.mqtt.proxy.impl.MQTTProxyException;
+import io.streamnative.pulsar.handlers.mqtt.proxy.web.WebService;
 import java.io.Closeable;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -74,6 +75,7 @@ public class MQTTProxyService implements Closeable {
     private Channel listenChannelTlsPsk;
     private final EventLoopGroup acceptorGroup;
     private final EventLoopGroup workerGroup;
+    private final WebService webService;
 
     private DefaultThreadFactory acceptorThreadFactory = new DefaultThreadFactory("mqtt-redirect-acceptor");
     private DefaultThreadFactory workerThreadFactory = new DefaultThreadFactory("mqtt-redirect-io");
@@ -101,7 +103,8 @@ public MQTTProxyService(BrokerService brokerService, MQTTProxyConfiguration prox
         this.eventService = proxyConfig.isSystemEventEnabled()
                 ? new SystemTopicBasedSystemEventService(pulsarService)
                 : new DisabledSystemEventService();
-        this.eventService.addListener(connectionManager.getEventListener());
+        this.eventService.addListener(connectionManager.getConnectListener());
+        this.eventService.addListener(connectionManager.getDisconnectListener());
         this.eventService.addListener(new RetainedMessageHandler(eventService).getEventListener());
         this.acceptorGroup = EventLoopUtil.newEventLoopGroup(proxyConfig.getMqttProxyNumAcceptorThreads(),
                 false, acceptorThreadFactory);
@@ -112,6 +115,7 @@ public MQTTProxyService(BrokerService brokerService, MQTTProxyConfiguration prox
         this.proxyAdapter = new MQTTProxyAdapter(this);
         this.sslContextRefresher = Executors.newSingleThreadScheduledExecutor(
                 new DefaultThreadFactory("mop-proxy-ssl-context-refresher"));
+        this.webService = new WebService(this);
     }
 
     private void configValid(MQTTProxyConfiguration proxyConfig) {
@@ -168,6 +172,7 @@ public void start() throws MQTTProxyException {
         }
         this.lookupHandler = new PulsarServiceLookupHandler(pulsarService, proxyConfig);
         this.eventService.start();
+        this.webService.start();
     }
 
     public void start0() throws MQTTProxyException {

diff --git a/...ava/io/streamnative/pulsar/handlers/mqtt/proxy/impl/MQTTProxyProtocolMethodProcessor.java b/...ava/io/streamnative/pulsar/handlers/mqtt/proxy/impl/MQTTProxyProtocolMethodProcessor.java
@@ -280,7 +280,11 @@ public void processDisconnect(final MqttAdapterMessage msg) {
     @Override
     public void processConnectionLost() {
         if (log.isDebugEnabled()) {
-            log.debug("[Proxy Connection Lost] [{}] ", connection.getClientId());
+            String clientId = "unknown";
+            if (connection != null) {
+                clientId = connection.getClientId();
+            }
+            log.debug("[Proxy Connection Lost] [{}] ", clientId);
         }
         autoSubscribeHandler.close();
         if (connection != null) {