anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add log traces

Browse Source
This commit is contained in:
Luis Barrios
2017-01-12 12:35:51 +01:00
parent 8b3f158a43
commit 03ae299e36
3 changed files with 126 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+29 -8
View File
@@ -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<StoredMessage> searchMatching(IMatchingCondition condition) {
LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size());
if (LOG.isDebugEnabled()) {
LOG.debug("Scanning retained messages...");
}
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, MessageGUID> 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<Integer, MessageGUID> 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<Integer, MessageGUID> 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);
}
}
broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java
+21 -6
View File
@@ -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.");
}
}
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+76 -26
View File
@@ -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<ClientTopicCouple> listAllSubscriptions() {
LOG.info("Retrieving existing subscriptions...");
final List<ClientTopicCouple> allSubscriptions = new ArrayList<>();
for (String clientID : m_persistentSessions.keySet()) {
ConcurrentMap<String, Subscription> 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<String, Subscription> 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<Subscription> getSubscriptions() {
LOG.info("Retrieving existing subscriptions...");
List<Subscription> subscriptions = new ArrayList<>();
for (String clientID : m_persistentSessions.keySet()) {
ConcurrentMap<String, Subscription> 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<Integer> 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<Integer, MessageGUID> 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<Integer, MessageGUID> m = this.m_inflightStore.get(clientID);
if (m == null) {
m = new ConcurrentHashMap<>();
@@ -209,20 +238,26 @@ class MapDBSessionsStore implements ISessionsStore {
@Override
public BlockingQueue<StoredMessage> 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<Integer, MessageGUID> 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<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
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<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap<Integer, MessageGUID>());
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<Integer, MessageGUID> 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<Integer, MessageGUID> 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);
}
}