From d7d97a74d8546cf917d568dce5267f7214eb65cb Mon Sep 17 00:00:00 2001 From: Luis Barrios Date: Thu, 12 Jan 2017 17:00:18 +0100 Subject: [PATCH] Add simple connection management capabilities to the broker. An external codebase will now be able to monitor and close the opened connections. --- .../connections/IConnectionsManager.java | 58 +++++++++++++ .../io/moquette/connections/MqttSession.java | 70 +++++++++++++++ .../connections/MqttSubscription.java | 38 ++++++++ .../server/ConnectionDescriptorStore.java | 86 +++++++++++++++++-- .../main/java/io/moquette/server/Server.java | 10 +++ .../java/io/moquette/spi/ClientSession.java | 17 ++++ .../java/io/moquette/spi/IMessagesStore.java | 2 + .../java/io/moquette/spi/ISessionsStore.java | 32 +++++++ .../moquette/spi/impl/ProtocolProcessor.java | 2 +- .../impl/ProtocolProcessorBootstrapper.java | 12 ++- .../spi/impl/subscriptions/Subscription.java | 4 + .../spi/persistence/MapDBMessagesStore.java | 7 ++ .../spi/persistence/MapDBSessionsStore.java | 30 +++++++ .../spi/impl/MemoryMessagesStore.java | 9 ++ .../spi/persistence/MemorySessionStore.java | 34 ++++++++ 15 files changed, 404 insertions(+), 7 deletions(-) create mode 100644 broker/src/main/java/io/moquette/connections/IConnectionsManager.java create mode 100644 broker/src/main/java/io/moquette/connections/MqttSession.java create mode 100644 broker/src/main/java/io/moquette/connections/MqttSubscription.java diff --git a/broker/src/main/java/io/moquette/connections/IConnectionsManager.java b/broker/src/main/java/io/moquette/connections/IConnectionsManager.java new file mode 100644 index 00000000..38d4b5ab --- /dev/null +++ b/broker/src/main/java/io/moquette/connections/IConnectionsManager.java @@ -0,0 +1,58 @@ +package io.moquette.connections; + +import java.util.Collection; + +/** + * This interface will be used by an external codebase to retrieve and close + * physical connections. + * + * @author lbarrios + * + */ +public interface IConnectionsManager { + /** + * Returns the number of physical connections + * + * @return + */ + public int getActiveConnectionsNo(); + + /** + * Determines wether a MQTT client is connected to the broker. + * + * @param clientID + * @return + */ + public boolean isConnected(String clientID); + + /** + * Returns the identifiers of the MQTT clients that are connected to the + * broker. + * + * @return + */ + public Collection getConnectedClientIds(); + + /** + * Closes a physical connection. + * + * @param clientID + * @param closeImmediately + * If false, the connection will be flushed before it is closed. + * @return + */ + public boolean closeConnection(String clientID, boolean closeImmediately); + + /** + * Returns the state of the session of a given client. + * @param clientID + * @return + */ + public MqttSession getSessionStatus(String clientID); + + /** + * Returns the state of all the sessions + * @return + */ + public Collection getSessions(); +} diff --git a/broker/src/main/java/io/moquette/connections/MqttSession.java b/broker/src/main/java/io/moquette/connections/MqttSession.java new file mode 100644 index 00000000..5d5082e9 --- /dev/null +++ b/broker/src/main/java/io/moquette/connections/MqttSession.java @@ -0,0 +1,70 @@ +package io.moquette.connections; + +import java.util.Collection; + +/** + * A class that represents the overall connection status of a MQTT session. Its + * instances will be used by an external codebase when the broker is configured + * in embedded mode. + * + * @author lbarrios + * + */ +public class MqttSession { + + private boolean connectionEstablished; + private boolean cleanSession; + private int inflightMessages; + private int pendingPublishMessagesNo; + private int secondPhaseAckPendingMessages; + private Collection activeSubscriptions; + + public boolean isConnectionEstablished() { + return connectionEstablished; + } + + public void setConnectionEstablished(boolean connectionEstablished) { + this.connectionEstablished = connectionEstablished; + } + + public boolean isCleanSession() { + return cleanSession; + } + + public void setCleanSession(boolean cleanSession) { + this.cleanSession = cleanSession; + } + + public int getPendingPublishMessagesNo() { + return pendingPublishMessagesNo; + } + + public void setPendingPublishMessagesNo(int pendingPublishMessagesNo) { + this.pendingPublishMessagesNo = pendingPublishMessagesNo; + } + + public int getInflightMessages() { + return inflightMessages; + } + + public void setInflightMessages(int inflightMessages) { + this.inflightMessages = inflightMessages; + } + + public int getSecondPhaseAckPendingMessages() { + return secondPhaseAckPendingMessages; + } + + public void setSecondPhaseAckPendingMessages(int secondPhaseAckPendingMessages) { + this.secondPhaseAckPendingMessages = secondPhaseAckPendingMessages; + } + + public Collection getActiveSubscriptions() { + return activeSubscriptions; + } + + public void setActiveSubscriptions(Collection activeSubscriptions) { + this.activeSubscriptions = activeSubscriptions; + } + +} diff --git a/broker/src/main/java/io/moquette/connections/MqttSubscription.java b/broker/src/main/java/io/moquette/connections/MqttSubscription.java new file mode 100644 index 00000000..1e47afa8 --- /dev/null +++ b/broker/src/main/java/io/moquette/connections/MqttSubscription.java @@ -0,0 +1,38 @@ +package io.moquette.connections; + +/** + * A class that represents a MQTT subscription. + * + * @author lbarrios + * + */ +public class MqttSubscription { + private final String requestedQos; + private final String clientId; + private final String topicFilter; + private final boolean active; + + public MqttSubscription(String requestedQos, String clientId, String topicFilter, boolean active) { + this.requestedQos = requestedQos; + this.clientId = clientId; + this.topicFilter = topicFilter; + this.active = active; + } + + public String getRequestedQos() { + return requestedQos; + } + + public String getClientId() { + return clientId; + } + + public String getTopicFilter() { + return topicFilter; + } + + public boolean isActive() { + return active; + } + +} diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java index 4030c292..732512ef 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java +++ b/broker/src/main/java/io/moquette/server/ConnectionDescriptorStore.java @@ -1,21 +1,31 @@ package io.moquette.server; +import java.util.ArrayList; +import java.util.Collection; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import io.moquette.connections.IConnectionsManager; +import io.moquette.connections.MqttSession; +import io.moquette.connections.MqttSubscription; import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.spi.ClientSession; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.impl.subscriptions.Subscription; -public class ConnectionDescriptorStore { +public class ConnectionDescriptorStore implements IConnectionsManager { private static final Logger LOG = LoggerFactory.getLogger(ConnectionDescriptorStore.class); private final ConcurrentMap connectionDescriptors; + private final ISessionsStore sessionsStore; - public ConnectionDescriptorStore() { + public ConnectionDescriptorStore(ISessionsStore sessionsStore) { this.connectionDescriptors = new ConcurrentHashMap(); + this.sessionsStore = sessionsStore; } public boolean sendMessage(AbstractMessage message, String clientId) { @@ -78,8 +88,74 @@ public class ConnectionDescriptorStore { return connectionDescriptors.get(clientID); } - public boolean isConnected(String clientID) { - return connectionDescriptors.containsKey(clientID); - } + @Override + public boolean isConnected(String clientID) { + return connectionDescriptors.containsKey(clientID); + } + + @Override + public int getActiveConnectionsNo() { + return connectionDescriptors.size(); + } + + @Override + public Collection getConnectedClientIds() { + return connectionDescriptors.keySet(); + } + + @Override + public boolean closeConnection(String clientID, boolean closeImmediately) { + ConnectionDescriptor descriptor = connectionDescriptors.get(clientID); + if (descriptor == null) { + LOG.error( + "The connection descriptor doesn't exist. The MQTT connection cannot be closed. MqttClientId = {}, closeImmediately = {}.", + clientID, closeImmediately); + return false; + } + if (closeImmediately) { + descriptor.abort(); + return true; + } else { + return descriptor.close(); + } + } + + @Override + public MqttSession getSessionStatus(String clientID) { + LOG.info("Retrieving status of session. MqttClientId = {}.", clientID); + ClientSession session = sessionsStore.sessionForClient(clientID); + if (session == null) { + LOG.error("The given MQTT client ID doesn't have an associated session. MqttClientId = {}.", clientID); + return null; + } + return buildMqttSession(session); + } + + @Override + public Collection getSessions() { + LOG.info("Retrieving status of all sessions."); + Collection result = new ArrayList(); + for (ClientSession session : sessionsStore.getAllSessions()) { + result.add(buildMqttSession(session)); + } + return result; + } + + private MqttSession buildMqttSession(ClientSession session) { + MqttSession result = new MqttSession(); + Collection mqttSubscriptions = new ArrayList<>(); + for (Subscription subscription : session.getSubscriptions()) { + mqttSubscriptions.add(new MqttSubscription(subscription.getRequestedQos().toString(), + subscription.getClientId(), subscription.getTopicFilter(), subscription.isActive())); + } + result.setActiveSubscriptions(mqttSubscriptions); + result.setCleanSession(session.isCleanSession()); + result.setConnectionEstablished(this.isConnected(session.clientID)); + result.setPendingPublishMessagesNo(session.getPendingPublishMessagesNo()); + result.setSecondPhaseAckPendingMessages(session.getSecondPhaseAckPendingMessages()); + result.setInflightMessages(session.getInflightMessagesNo()); + LOG.info("The status of the session has been retrieved successfully. MqttClientId = {}.", session.clientID); + return result; + } } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index dfcb2194..af523bb8 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -23,6 +23,7 @@ import com.hazelcast.core.HazelcastInstance; import com.hazelcast.core.HazelcastInstanceNotActiveException; import com.hazelcast.core.ITopic; import io.moquette.BrokerConstants; +import io.moquette.connections.IConnectionsManager; import io.moquette.interception.HazelcastInterceptHandler; import io.moquette.interception.HazelcastMsg; import io.moquette.interception.InterceptHandler; @@ -295,4 +296,13 @@ public class Server { m_processor.removeInterceptHandler(interceptHandler); } + /** + * Returns the connections manager of this broker. + * + * @return + */ + public IConnectionsManager getConnectionsManager() { + return m_processorBootstrapper.getConnectionDescriptors(); + } + } diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 6b83c053..b4179b8e 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -189,4 +189,21 @@ public class ClientSession { public IMessagesStore.StoredMessage getInflightMessage(int messageID) { return m_sessionsStore.getInflightMessage(clientID, messageID); } + + public Set getSubscriptions() { + return subscriptions; + } + + public int getPendingPublishMessagesNo() { + return m_sessionsStore.getPendingPublishMessagesNo(clientID); + } + + public int getSecondPhaseAckPendingMessages() { + return m_sessionsStore.getSecondPhaseAckPendingMessages(clientID); + } + + public int getInflightMessagesNo() { + return m_sessionsStore.getInflightMessagesNo(clientID); + } + } \ No newline at end of file diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index f4d64579..03234374 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -135,4 +135,6 @@ public interface IMessagesStore { StoredMessage getMessageByGuid(MessageGUID guid); void cleanRetained(String topic); + + int getPendingPublishMessages(String clientID); } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 006ab0f9..422b02f3 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -15,6 +15,7 @@ */ package io.moquette.spi; +import java.util.Collection; import java.util.List; import java.util.concurrent.BlockingQueue; @@ -117,6 +118,12 @@ public interface ISessionsStore { * @return the session for the given clientID, null if not found. * */ ClientSession sessionForClient(String clientID); + + /** + * Returns all the sessions + * @return + */ + Collection getAllSessions(); void inFlightAck(String clientID, int messageID); @@ -156,4 +163,29 @@ public interface ISessionsStore { MessageGUID mapToGuid(String clientID, int messageID); StoredMessage getInflightMessage(String clientID, int messageID); + + /** + * Returns the number of inflight messages for the given client ID + * + * @param clientID + * @return + */ + int getInflightMessagesNo(String clientID); + + /** + * Returns the size of the session queue for the given client ID + * + * @param clientID + * @return + */ + int getPendingPublishMessagesNo(String clientID); + + /** + * Returns the number of second-phase ACK pending messages for the given + * client ID + * + * @param clientID + * @return + */ + int getSecondPhaseAckPendingMessages(String clientID); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 3767f6da..69eec972 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -164,7 +164,7 @@ public class ProtocolProcessor { IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { - init(new ConnectionDescriptorStore(), subscriptions, storageService, sessionsStore, authenticator, + init(new ConnectionDescriptorStore(sessionsStore), subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, serverPort); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index cf0358e8..e2dfeb11 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -17,6 +17,7 @@ package io.moquette.spi.impl; import io.moquette.BrokerConstants; import io.moquette.interception.InterceptHandler; +import io.moquette.server.ConnectionDescriptorStore; import io.moquette.server.Server; import io.moquette.server.config.IConfig; import io.moquette.server.config.IResourceLoader; @@ -59,6 +60,8 @@ public class ProtocolProcessorBootstrapper { private BrokerInterceptor m_interceptor; private final ProtocolProcessor m_processor = new ProtocolProcessor(); + + private ConnectionDescriptorStore connectionDescriptors; public ProtocolProcessorBootstrapper() { } @@ -141,11 +144,14 @@ public class ProtocolProcessorBootstrapper { } LOG.info("An {} authorizator instance will be used.", authorizator.getClass().getName()); } + + LOG.info("Initializing connection descriptor store..."); + connectionDescriptors = new ConnectionDescriptorStore(m_sessionsStore); LOG.info("Initializing MQTT protocol processor..."); boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); boolean allowZeroByteClientId = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false")); - m_processor.init(subscriptions, messagesStore, m_sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); + m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); return m_processor; } @@ -214,4 +220,8 @@ public class ProtocolProcessorBootstrapper { public void shutdown() { this.m_mapStorage.close(); } + + public ConnectionDescriptorStore getConnectionDescriptors() { + return connectionDescriptors; + } } diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java index b088b6ab..0f059fca 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java @@ -60,6 +60,10 @@ public final class Subscription implements Serializable { return topicFilter; } + public boolean isActive() { + return active; + } + @Override public boolean equals(Object o) { if (this == o) return true; diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java index eb039001..b4418afc 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -137,4 +137,11 @@ class MapDBMessagesStore implements IMessagesStore { } m_retainedStore.remove(topic); } + + @Override + public int getPendingPublishMessages(String clientID) { + ConcurrentMap messageIdToGuidMap = m_db + .getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)); + return messageIdToGuidMap.size(); + } } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index 012590c7..6db1f306 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -168,6 +168,15 @@ class MapDBSessionsStore implements ISessionsStore { PersistentSession storedSession = m_persistentSessions.get(clientID); return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); } + + @Override + public Collection getAllSessions() { + Collection result = new ArrayList(); + for (Map.Entry entry : m_persistentSessions.entrySet()) { + result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); + } + return result; + } @Override public void updateCleanStatus(String clientID, boolean cleanSession) { @@ -318,4 +327,25 @@ class MapDBSessionsStore implements ISessionsStore { } return m_messagesStore.getMessageByGuid(guid); } + + @Override + public int getInflightMessagesNo(String clientID) { + if (!m_inflightStore.containsKey(clientID)) + return 0; + else + return m_inflightStore.get(clientID).size(); + } + + @Override + public int getPendingPublishMessagesNo(String clientID) { + return m_messagesStore.getPendingPublishMessages(clientID); + } + + @Override + public int getSecondPhaseAckPendingMessages(String clientID) { + if (!m_secondPhaseStore.containsKey(clientID)) + return 0; + else + return m_secondPhaseStore.get(clientID).size(); + } } diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java index cad1d230..968d9381 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java @@ -98,4 +98,13 @@ public class MemoryMessagesStore implements IMessagesStore { public void cleanRetained(String topic) { m_retainedStore.remove(topic); } + + @Override + public int getPendingPublishMessages(String clientID) { + Map messageToGuids = m_messageToGuids.get(clientID); + if (messageToGuids == null) + return 0; + else + return messageToGuids.size(); + } } diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index aefd83fb..4de2ede2 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -23,6 +23,7 @@ import io.moquette.spi.ISessionsStore; import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.Utils; import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -129,6 +130,15 @@ public class MemorySessionStore implements ISessionsStore { MapDBPersistentStore.PersistentSession storedSession = m_persistentSessions.get(clientID); return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); } + + @Override + public Collection getAllSessions() { + Collection result = new ArrayList(); + for (Map.Entry entry : m_persistentSessions.entrySet()) { + result.add(new ClientSession(entry.getKey(), m_messagesStore, this, entry.getValue().cleanSession)); + } + return result; + } @Override public void updateCleanStatus(String clientID, boolean cleanSession) { @@ -252,4 +262,28 @@ public class MemorySessionStore implements ISessionsStore { public StoredMessage getInflightMessage( String clientID, int messageID ) { return null; } + + @Override + public int getInflightMessagesNo(String clientID) { + Map inflightMessages = m_inflightStore.get(clientID); + if (inflightMessages == null) + return 0; + else + return inflightMessages.size(); + } + + @Override + public int getPendingPublishMessagesNo(String clientID) { + return m_messagesStore.getPendingPublishMessages(clientID); + } + + @Override + public int getSecondPhaseAckPendingMessages(String clientID) { + Map pendingAcks = m_secondPhaseStore.get(clientID); + if (pendingAcks == null) + return 0; + else + return pendingAcks.size(); + } + }