From b827ad446d4f6f17d5240f0c4756861dbc795619 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 24 Feb 2019 18:15:49 +0100 Subject: [PATCH] Exposed method to retrieve the list of Server's connected clients (issue #448) --- .../io/moquette/broker/ClientDescriptor.java | 52 +++++++++++++++++++ .../io/moquette/broker/MQTTConnection.java | 5 ++ .../broker/MQTTConnectionFactory.java | 2 +- .../main/java/io/moquette/broker/Server.java | 29 +++++------ .../main/java/io/moquette/broker/Session.java | 8 +++ .../io/moquette/broker/SessionRegistry.java | 19 +++++++ 6 files changed, 97 insertions(+), 18 deletions(-) create mode 100644 broker/src/main/java/io/moquette/broker/ClientDescriptor.java diff --git a/broker/src/main/java/io/moquette/broker/ClientDescriptor.java b/broker/src/main/java/io/moquette/broker/ClientDescriptor.java new file mode 100644 index 00000000..6cca2f63 --- /dev/null +++ b/broker/src/main/java/io/moquette/broker/ClientDescriptor.java @@ -0,0 +1,52 @@ +package io.moquette.broker; + +import java.util.Objects; + +public class ClientDescriptor { + + private final String clientID; + private final String address; + private final int port; + + ClientDescriptor(String clientID, String address, int port) { + this.clientID = clientID; + this.address = address; + this.port = port; + } + + public String getClientID() { + return clientID; + } + + public String getAddress() { + return address; + } + + public int getPort() { + return port; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + ClientDescriptor that = (ClientDescriptor) o; + return port == that.port && + Objects.equals(clientID, that.clientID) && + Objects.equals(address, that.address); + } + + @Override + public int hashCode() { + return Objects.hash(clientID, address, port); + } + + @Override + public String toString() { + return "ClientDescriptor{" + + "clientID='" + clientID + '\'' + + ", address='" + address + '\'' + + ", port=" + port + + '}'; + } +} diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java index 934b3a55..5e6199d6 100644 --- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java +++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java @@ -25,6 +25,7 @@ import io.netty.handler.timeout.IdleStateHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import java.util.*; import java.util.concurrent.TimeUnit; @@ -488,4 +489,8 @@ final class MQTTConnection { public String toString() { return "MQTTConnection{channel=" + channel + ", connected=" + connected + '}'; } + + InetSocketAddress remoteAddress() { + return (InetSocketAddress) channel.remoteAddress(); + } } diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java b/broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java index 6321cd8d..f171dcc6 100644 --- a/broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java +++ b/broker/src/main/java/io/moquette/broker/MQTTConnectionFactory.java @@ -33,7 +33,7 @@ class MQTTConnectionFactory { this.postOffice = postOffice; } - public MQTTConnection create(Channel channel) { + MQTTConnection create(Channel channel) { return new MQTTConnection(channel, brokerConfig, authenticator, sessionRegistry, postOffice); } } diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java index 2d2ecee8..8f1ec825 100644 --- a/broker/src/main/java/io/moquette/broker/Server.java +++ b/broker/src/main/java/io/moquette/broker/Server.java @@ -34,10 +34,7 @@ import java.io.File; import java.io.IOException; import java.lang.reflect.InvocationTargetException; import java.text.ParseException; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Properties; +import java.util.*; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -53,6 +50,7 @@ public class Server { private PostOffice dispatcher; private BrokerInterceptor interceptor; private H2Builder h2Builder; + private SessionRegistry sessions; public static void main(String[] args) throws IOException { final Server server = new Server(); @@ -136,7 +134,7 @@ public class Server { } public void startServer(IConfig config, List handlers, ISslContextCreator sslCtxCreator, - IAuthenticator authenticator, IAuthorizatorPolicy authorizatorPolicy) throws IOException { + IAuthenticator authenticator, IAuthorizatorPolicy authorizatorPolicy) { final long start = System.currentTimeMillis(); if (handlers == null) { handlers = Collections.emptyList(); @@ -179,7 +177,7 @@ public class Server { ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); subscriptions.init(subscriptionsRepository); final Authorizator authorizator = new Authorizator(authorizatorPolicy); - SessionRegistry sessions = new SessionRegistry(subscriptions, queueRepository, authorizator); + sessions = new SessionRegistry(subscriptions, queueRepository, authorizator); dispatcher = new PostOffice(subscriptions, retainedRepository, sessions, interceptor, authorizator); final BrokerConfiguration brokerConfig = new BrokerConfiguration(config); MQTTConnectionFactory connectionFactory = new MQTTConnectionFactory(brokerConfig, authenticator, sessions, @@ -195,11 +193,6 @@ public class Server { } private IAuthorizatorPolicy initializeAuthorizatorPolicy(IAuthorizatorPolicy authorizatorPolicy, IConfig props) { -// if (authorizatorPolicy == null) { -// authorizatorPolicy = new PermitAllAuthorizatorPolicy(); -// } -// return authorizatorPolicy; - LOG.debug("Configuring MQTT authorizator policy"); String authorizatorClassName = props.getProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, ""); if (authorizatorPolicy == null && !authorizatorClassName.isEmpty()) { @@ -215,7 +208,7 @@ public class Server { IResourceLoader resourceLoader = props.getResourceLoader(); authorizatorPolicy = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath)); } catch (ParseException pex) { - LOG.error("Unable to parse ACL file. path=" + aclFilePath, pex); + LOG.error("Unable to parse ACL file. path = {}", aclFilePath, pex); } } else { authorizatorPolicy = new PermitAllAuthorizatorPolicy(); @@ -226,11 +219,6 @@ public class Server { } private IAuthenticator initializeAuthenticator(IAuthenticator authenticator, IConfig props) { -// if (authenticator == null) { -// authenticator = new AcceptAllAuthenticator(); -// } -// return authenticator; - LOG.debug("Configuring MQTT authenticator"); String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, ""); @@ -391,4 +379,11 @@ public class Server { LOG.info("Removing MQTT message interceptor. InterceptorId={}", interceptHandler.getID()); interceptor.removeInterceptHandler(interceptHandler); } + + /** + * Return a list of descriptors of connected clients. + * */ + public Collection listConnectedClients() { + return sessions.listConnectedClients(); + } } diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java index 35622184..f914cdab 100644 --- a/broker/src/main/java/io/moquette/broker/Session.java +++ b/broker/src/main/java/io/moquette/broker/Session.java @@ -23,6 +23,7 @@ import io.netty.util.ReferenceCountUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.net.InetSocketAddress; import java.util.*; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; @@ -368,6 +369,13 @@ class Session { ReferenceCountUtil.release(removedMsg); } + Optional remoteAddress() { + if (connected()) { + return Optional.of(mqttConnection.remoteAddress()); + } + return Optional.empty(); + } + @Override public String toString() { return "Session{" + diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java index fbd16229..d6d26d68 100644 --- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java +++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java @@ -26,9 +26,13 @@ import io.netty.handler.codec.mqtt.MqttQoS; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.net.InetSocketAddress; +import java.util.Collection; +import java.util.Optional; import java.util.Queue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.stream.Collectors; public class SessionRegistry { @@ -245,4 +249,19 @@ public class SessionRegistry { private void dropQueuesForClient(String clientId) { queues.remove(clientId); } + + Collection listConnectedClients() { + return pool.values().stream() + .filter(Session::connected) + .map(this::createClientDescriptor) + .filter(Optional::isPresent) + .map(Optional::get) + .collect(Collectors.toList()); + } + + private Optional createClientDescriptor(Session s) { + final String clientID = s.getClientID(); + final Optional remoteAddressOpt = s.remoteAddress(); + return remoteAddressOpt.map(r -> new ClientDescriptor(clientID, r.getHostString(), r.getPort())); + } }