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 62f29275..eb039001 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -48,18 +48,24 @@ class MapDBMessagesStore implements IMessagesStore { @Override public void initStore() { + LOG.info("Initializing store..."); m_retainedStore = m_db.getHashMap("retained"); m_persistentMessageStore = m_db.getHashMap("persistedMessages"); } @Override public void storeRetained(String topic, MessageGUID guid) { + if (LOG.isDebugEnabled()) { + LOG.debug("Storing retained messages. Topic = {}, guid = {}.", topic, guid); + } m_retainedStore.put(topic, guid); } @Override public Collection searchMatching(IMatchingCondition condition) { - LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); + if (LOG.isDebugEnabled()) { + LOG.debug("Scanning retained messages..."); + } List results = new ArrayList<>(); for (Map.Entry entry : m_retainedStore.entrySet()) { @@ -70,19 +76,22 @@ class MapDBMessagesStore implements IMessagesStore { } } + if (LOG.isDebugEnabled()) { + LOG.debug("The retained messages have been scanned. MatchingMessages = {}.", results); + } + return results; } @Override public MessageGUID storePublishForFuture(StoredMessage storedMessage) { - LOG.debug("storePublishForFuture store evt {}", storedMessage); - if (storedMessage.getClientID() == null) { - LOG.error("persisting a message without a clientID, bad programming error msg: {}", storedMessage); - throw new IllegalArgumentException("persisting a message without a clientID, bad programming error"); - } + assert storedMessage.getClientID() != null : "The message to be persisted must have a valid client ID"; MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); storedMessage.setGuid(guid); - LOG.debug("storePublishForFuture guid <{}>", guid); + if (LOG.isDebugEnabled()) { + LOG.debug("Storing publish event. MqttClientId = {}, messageId = {}, guid = {}, topic = {}.", storedMessage.getClientID(), + storedMessage.getMessageID(), guid, storedMessage.getTopic()); + } m_persistentMessageStore.put(guid, storedMessage); ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(storedMessage.getClientID())); messageIdToGuid.put(storedMessage.getMessageID(), guid); @@ -91,6 +100,9 @@ class MapDBMessagesStore implements IMessagesStore { @Override public void dropMessagesInSession(String clientID) { + if (LOG.isDebugEnabled()) { + LOG.debug("Dropping stored messages. ClientId = {}.", clientID); + } ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)); for (MessageGUID guid : messageIdToGuid.values()) { removeStoredMessage(guid); @@ -102,18 +114,27 @@ class MapDBMessagesStore implements IMessagesStore { //remove only the not retained and no more referenced StoredMessage storedMessage = m_persistentMessageStore.get(guid); if (!storedMessage.isRetained()) { - LOG.debug("Cleaning not retained message guid {}", guid); + if (LOG.isDebugEnabled()) { + LOG.debug("Dropping stored message. ClientId = {}, messageId = {}, guid = {}, topic = {}.", + storedMessage.getClientID(), storedMessage.getMessageID(), guid, storedMessage.getTopic()); + } m_persistentMessageStore.remove(guid); } } @Override public StoredMessage getMessageByGuid(MessageGUID guid) { + if (LOG.isDebugEnabled()) { + LOG.debug("Retrieving stored message. Guid = {}.", guid); + } return m_persistentMessageStore.get(guid); } @Override public void cleanRetained(String topic) { + if (LOG.isDebugEnabled()) { + LOG.debug("Cleaning retained messages. Topic = {}.", topic); + } m_retainedStore.remove(topic); } } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java index 31058a5c..b8a39db0 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java @@ -81,23 +81,30 @@ public class MapDBPersistentStore { } public void initStore() { + LOG.info("Initializing MapDB store..."); if (m_storePath == null || m_storePath.isEmpty()) { + LOG.warn("The MapDB store file path is empty. Using in-memory store."); m_db = DBMaker.newMemoryDB().make(); } else { File tmpFile; try { + LOG.info("Using user-defined MapDB store file. Path = {}.", m_storePath); tmpFile = new File(m_storePath); boolean fileNewlyCreated = tmpFile.createNewFile(); - LOG.info("Starting with {} [{}] db file", fileNewlyCreated ? "fresh" : "existing", m_storePath); + LOG.warn("Using {} MapDB store file. Path = {}.", fileNewlyCreated ? "fresh" : "existing", m_storePath); } catch (IOException ex) { - LOG.error(null, ex); + LOG.error("Unable to open MapDB store file. Path = {}, cause = {}, errorMessage = {}.", m_storePath, ex.getCause(), ex.getMessage()); throw new MQTTException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex); } m_db = DBMaker.newFileDB(tmpFile).make(); } + LOG.info("Scheduling MapDB commit task..."); m_scheduler.scheduleWithFixedDelay(new Runnable() { @Override public void run() { + if (LOG.isDebugEnabled()) { + LOG.debug("Committing to MapDB..."); + } m_db.commit(); } }, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS); @@ -112,14 +119,22 @@ public class MapDBPersistentStore { public void close() { if (this.m_db.isClosed()) { - LOG.debug("already closed"); + LOG.warn("The MapDB store is already closed. Nothing will be done."); return; } + LOG.info("Performing last commit to MapDB..."); this.m_db.commit(); - //LOG.debug("persisted subscriptions {}", m_persistentSubscriptions); + LOG.info("Closing MapDB store..."); this.m_db.close(); - LOG.debug("closed disk storage"); + LOG.info("Stopping MapDB commit tasks..."); this.m_scheduler.shutdown(); - LOG.debug("Persistence commit scheduler is shutdown"); + try { + m_scheduler.awaitTermination(10L, TimeUnit.SECONDS); + } catch (InterruptedException e) {} + if (!m_scheduler.isTerminated()) { + LOG.warn("Forcing shutdown of MapDB commit tasks..."); + m_scheduler.shutdown(); + } + LOG.info("The MapDB store has been closed successfully."); } } 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 38515931..012590c7 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -66,38 +66,44 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void addNewSubscription(Subscription newSubscription) { - LOG.debug("addNewSubscription invoked with subscription {}", newSubscription); + LOG.info("Adding new subscription. MqttClientId = {}, topics = {}.", newSubscription.getClientId(), + newSubscription.getTopicFilter()); final String clientID = newSubscription.getClientId(); m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); - if (LOG.isTraceEnabled()) { - LOG.trace("subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); + if (LOG.isDebugEnabled()) { + LOG.debug("The subscription has been added. MqttClientId = {}, topics = {}, clientSubscriptions = {}.", + newSubscription.getClientId(), newSubscription.getTopicFilter(), + m_db.getHashMap("subscriptions_" + clientID)); } } @Override public void removeSubscription(String topicFilter, String clientID) { - LOG.debug("removeSubscription topic filter: {} for clientID: {}", topicFilter, clientID); + LOG.info("Removing subscription. MqttClientId = {}, topics = {}.", clientID, topicFilter); if (!m_db.exists("subscriptions_" + clientID)) { return; } m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); + if (LOG.isDebugEnabled()) { + LOG.debug("The subscription has been removed. MqttClientId = {}, topics = {}, clientSubscriptions = {}.", + clientID, topicFilter, m_db.getHashMap("subscriptions_" + clientID)); + } } @Override public void wipeSubscriptions(String clientID) { - LOG.debug("wipeSubscriptions"); - if (LOG.isTraceEnabled()) { - LOG.trace("Subscription pre wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); - } + LOG.info("Wiping subscriptions. MqttClientId = {}.", clientID); m_db.delete("subscriptions_" + clientID); - if (LOG.isTraceEnabled()) { - LOG.trace("Subscription post wipe: subscriptions_{}: {}", clientID, m_db.getHashMap("subscriptions_" + clientID)); + if (LOG.isDebugEnabled()) { + LOG.debug("The subscriptions have been removed. MqttClientId = {}, clientSubscriptions = {}.", clientID, + m_db.getHashMap("subscriptions_" + clientID)); } } @Override public List listAllSubscriptions() { + LOG.info("Retrieving existing subscriptions..."); final List allSubscriptions = new ArrayList<>(); for (String clientID : m_persistentSessions.keySet()) { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); @@ -105,24 +111,31 @@ class MapDBSessionsStore implements ISessionsStore { allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter)); } } - LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions); + if (LOG.isDebugEnabled()) { + LOG.debug("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions); + } return allSubscriptions; } @Override public Subscription getSubscription(ClientTopicCouple couple) { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + couple.clientID); - LOG.debug("subscriptions_{}: {}", couple.clientID, clientSubscriptions); + LOG.info("Retrieving subscriptions. MqttClientId = {}, subscriptions = {}.", couple.clientID, + clientSubscriptions); return clientSubscriptions.get(couple.topicFilter); } @Override public List getSubscriptions() { + LOG.info("Retrieving existing subscriptions..."); List subscriptions = new ArrayList<>(); for (String clientID : m_persistentSessions.keySet()) { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); subscriptions.addAll(clientSubscriptions.values()); } + if (LOG.isDebugEnabled()) { + LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions); + } return subscriptions; } @@ -133,19 +146,22 @@ class MapDBSessionsStore implements ISessionsStore { @Override public ClientSession createNewSession(String clientID, boolean cleanSession) { - LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession); if (m_persistentSessions.containsKey(clientID)) { - LOG.error("already exists a session for client <{}>, bad condition", clientID); + LOG.error( + "Unable to create a new session: the client ID is already in use. MqttClientId = {}, cleanSession = {}.", + clientID, cleanSession); throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); } - LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); + LOG.info("Creating new session. MqttClientId = {}, cleanSession = {}.", clientID, cleanSession); m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession)); return new ClientSession(clientID, m_messagesStore, this, cleanSession); } @Override public ClientSession sessionForClient(String clientID) { + LOG.info("Retrieving session. MqttClientId = {}.", clientID); if (!m_persistentSessions.containsKey(clientID)) { + LOG.warn("The session does not exist. MqttClientId = {}.", clientID); return null; } @@ -155,6 +171,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void updateCleanStatus(String clientID, boolean cleanSession) { + LOG.info("Updating cleanSession flag. MqttClientId = {}, cleanSession = {}.", clientID, cleanSession); m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession)); } @@ -163,6 +180,9 @@ class MapDBSessionsStore implements ISessionsStore { * */ @Override public int nextPacketID(String clientID) { + if (LOG.isDebugEnabled()) { + LOG.debug("Generating next packet ID. MqttClientId = {}.", clientID); + } Set inFlightForClient = this.m_inFlightIds.get(clientID); if (inFlightForClient == null) { int nextPacketId = 1; @@ -176,15 +196,20 @@ class MapDBSessionsStore implements ISessionsStore { int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); int nextPacketId = (maxId + 1) % 0xFFFF; inFlightForClient.add(nextPacketId); + if (LOG.isDebugEnabled()) { + LOG.debug("The next packet ID has been generated. MqttClientId = {}, result = {}.", clientID, nextPacketId); + } return nextPacketId; } @Override public void inFlightAck(String clientID, int messageID) { - LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); + if (LOG.isDebugEnabled()) { + LOG.debug("Acknowledging inflight message. MqttClientId = {}, messageId = {}.", clientID, messageID); + } Map m = this.m_inflightStore.get(clientID); if (m == null) { - LOG.error("Can't find the inFlight record for client <{}>", clientID); + LOG.warn("Unable to retrieve inflight message record. MqttClientId = {}, messageId = {}.", clientID, messageID); return; } m.remove(messageID); @@ -198,6 +223,10 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void inFlight(String clientID, int messageID, MessageGUID guid) { + if (LOG.isDebugEnabled()) { + LOG.debug("Storing inflight message. MqttClientId = {}, messageId = {}, guid = {}.", clientID, messageID, + guid); + } ConcurrentMap m = this.m_inflightStore.get(clientID); if (m == null) { m = new ConcurrentHashMap<>(); @@ -209,20 +238,26 @@ class MapDBSessionsStore implements ISessionsStore { @Override public BlockingQueue queue(String clientID) { + LOG.info("Queuing pending message. MqttClientId = {}, guid = {}.", clientID); return this.m_db.getQueue(clientID); } @Override public void dropQueue(String clientID) { + LOG.info("Removing pending messages. MqttClientId = {}.", clientID); this.m_db.delete(clientID); } @Override public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { - LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); + if (LOG.isDebugEnabled()) { + LOG.debug("Moving inflight message to second phase ack state. MqttClientId = {}, messageID = {}.", clientID, + messageID); + } Map m = this.m_inflightStore.get(clientID); if (m == null) { - LOG.error("Can't find the inFlight record for client <{}>", clientID); + LOG.warn("Unable to retrieve inflight message record. MqttClientId = {}, messageId = {}.", clientID, + messageID); return; } MessageGUID guid = m.remove(messageID); @@ -233,7 +268,6 @@ class MapDBSessionsStore implements ISessionsStore { inFlightForClient.remove(messageID); } - LOG.info("Moving to second phase store"); Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); messageIDs.put(messageID, guid); m_secondPhaseStore.put(clientID, messageIDs); @@ -241,6 +275,9 @@ class MapDBSessionsStore implements ISessionsStore { @Override public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { + if (LOG.isDebugEnabled()) { + LOG.debug("Processing second phase ACK. MqttClientId = {}, messageId = {}.", clientID, messageID); + } Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); MessageGUID guid = messageIDs.remove(messageID); m_secondPhaseStore.put(clientID, messageIDs); @@ -249,8 +286,16 @@ class MapDBSessionsStore implements ISessionsStore { @Override public MessageGUID mapToGuid(String clientID, int messageID) { + if (LOG.isDebugEnabled()) { + LOG.debug("Mapping message ID to GUID. MqttClientId = {}, messageId = {}.", clientID, messageID); + } ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); - return messageIdToGuid.get(messageID); + MessageGUID result = messageIdToGuid.get(messageID); + if (LOG.isDebugEnabled()) { + LOG.debug("The message ID has been mapped to a GUID. MqttClientId = {}, messageId = {}, guid = {}.", + clientID, messageID, result); + } + return result; } static String messageId2GuidsMapName(String clientID) { @@ -259,13 +304,18 @@ class MapDBSessionsStore implements ISessionsStore { @Override public StoredMessage getInflightMessage(String clientID, int messageID) { + LOG.info("Retrieving inflight message. MqttClientId = {}, messageId = {}.", clientID, messageID); Map clientEntries = m_inflightStore.get(clientID); - if (clientEntries == null) - return null; + if (clientEntries == null) { + LOG.warn("The client has no inflight messages. MqttClientId = {}, messageId = {}.", clientID, messageID); + return null; + } MessageGUID guid = clientEntries.get(messageID); - LOG.info("inflight messageID {} guid <{}>", messageID, guid); - if (guid == null) - return null; + if (guid == null) { + LOG.warn("The message ID does not have an associated GUID. MqttClientId = {}, messageId = {}.", clientID, + messageID); + return null; + } return m_messagesStore.getMessageByGuid(guid); } }