From 716c9ac0a061f144d672ec020de5e5d2ea8b419e Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 18 Jan 2017 23:27:02 +0100 Subject: [PATCH] Minor on logging, removed ugly if(isDebugEnabled()) for debug logging, kept only when the toString generation could potentially take long, and trasformed in tracing logs. Moved Error to Warn log level --- .../main/java/io/moquette/server/Server.java | 10 ++--- .../server/netty/NettyMQTTHandler.java | 4 +- .../java/io/moquette/spi/ClientSession.java | 2 +- .../moquette/spi/impl/BrokerInterceptor.java | 43 +++++++------------ .../spi/impl/InternalRepublisher.java | 12 ++---- .../moquette/spi/impl/MessagesPublisher.java | 15 +++---- .../moquette/spi/impl/ProtocolProcessor.java | 12 ++---- .../moquette/spi/impl/Qos0PublishHandler.java | 4 +- .../moquette/spi/impl/Qos1PublishHandler.java | 4 +- .../moquette/spi/impl/Qos2PublishHandler.java | 12 ++---- .../subscriptions/SubscriptionsStore.java | 14 +++--- .../spi/persistence/MapDBMessagesStore.java | 37 +++++----------- .../spi/persistence/MapDBPersistentStore.java | 4 +- .../spi/persistence/MapDBSessionsStore.java | 31 +++++-------- .../netty/performance/LoopMQTTHandler.java | 4 +- .../NettyPublishReceiverHandler.java | 4 +- .../performance/PublishReceiverHandler.java | 4 +- 17 files changed, 72 insertions(+), 144 deletions(-) diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index bb1e2f7e..dbbd0ad0 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -225,13 +225,11 @@ public class Server { * */ public void internalPublish(PublishMessage msg) { if (!m_initialized) { - LOG.error("The server is not started. The message cannot be published. MqttClientId = {}, messageId = {}.", - msg.getClientId(), msg.getMessageID()); - throw new IllegalStateException("Can't publish on a server is not yet started"); - } - if (LOG.isDebugEnabled()) { - LOG.debug("Publishing message. MqttClientId = {}, messageId = {}.", msg.getClientId(), msg.getMessageID()); + LOG.error("The server is not started. The message cannot be published. MqttClientId = {}, messageId = {}.", + msg.getClientId(), msg.getMessageID()); + throw new IllegalStateException("Can't publish on a server is not yet started"); } + LOG.debug("Publishing message. MqttClientId = {}, messageId = {}.", msg.getClientId(), msg.getMessageID()); m_processor.internalPublish(msg); } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index 0e4a9854..c97c5d45 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -46,9 +46,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { public void channelRead(ChannelHandlerContext ctx, Object message) { AbstractMessage msg = (AbstractMessage) message; String messageType = Utils.msgType2String(msg.getMessageType()); - if (LOG.isDebugEnabled()) { - LOG.debug("Processing MQTT message. MessageType = {}.", messageType); - } + LOG.debug("Processing MQTT message. MessageType = {}.", messageType); try { switch (msg.getMessageType()) { case CONNECT: diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index b4179b8e..a29c635e 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -88,7 +88,7 @@ public class ClientSession { newSubscription.getTopicFilter(), AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())); boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter()); if (!validTopic) { - LOG.error("The topic filter is not valid. MqttClientId = {}, topics = {}.", newSubscription.getClientId(), + LOG.warn("The topic filter is not valid. MqttClientId = {}, topics = {}.", newSubscription.getClientId(), newSubscription.getTopicFilter()); //send SUBACK with 0x80 for this topic filter return false; diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index 5a49fae3..61c2d06d 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -100,10 +100,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientConnected(final ConnectMessage msg) { for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) { - if (LOG.isDebugEnabled()) { - LOG.debug("Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.", - msg.getClientID(), handler.getID()); - } + LOG.debug("Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.", + msg.getClientID(), handler.getID()); executor.execute(new Runnable() { @Override public void run() { @@ -116,10 +114,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientDisconnected(final String clientID, final String username ) { for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) { - if (LOG.isDebugEnabled()) { - LOG.debug("Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, interceptorId = {}.", - clientID, username, handler.getID()); - } + LOG.debug("Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, " + + "interceptorId = {}.", clientID, username, handler.getID()); executor.execute(new Runnable() { @Override public void run() { @@ -132,10 +128,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientConnectionLost(final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) { - if (LOG.isDebugEnabled()) { - LOG.debug("Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, interceptorId = {}.", - clientID, username, handler.getID()); - } + LOG.debug("Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, " + + "username = {}, interceptorId = {}.", clientID, username, handler.getID()); executor.execute(new Runnable() { @Override public void run() { @@ -148,10 +142,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicPublished(final PublishMessage msg, final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) { - if (LOG.isDebugEnabled()) { - LOG.debug("Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, interceptorId = {}.", - clientID, msg.getMessageID(), msg.getTopicName(), handler.getID()); - } + LOG.debug("Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " + + "interceptorId = {}.", clientID, msg.getMessageID(), msg.getTopicName(), handler.getID()); executor.execute(new Runnable() { @Override public void run() { @@ -164,10 +156,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicSubscribed(final Subscription sub, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) { - if (LOG.isDebugEnabled()) { - LOG.debug("Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, interceptorId = {}.", - sub.getClientId(), sub.getTopicFilter(), handler.getID()); - } + LOG.debug("Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, " + + "interceptorId = {}.", sub.getClientId(), sub.getTopicFilter(), handler.getID()); executor.execute(new Runnable() { @Override public void run() { @@ -180,10 +170,8 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) { - if (LOG.isDebugEnabled()) { - LOG.debug("Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, interceptorId = {}.", - clientID, topic, handler.getID()); - } + LOG.debug("Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, " + + "interceptorId = {}.", clientID, topic, handler.getID()); executor.execute(new Runnable() { @Override public void run() { @@ -196,10 +184,9 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyMessageAcknowledged( final InterceptAcknowledgedMessage msg ) { for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) { - if (LOG.isDebugEnabled()) { - LOG.debug("Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, interceptorId = {}.", - msg.getMsg().getClientID(), msg.getMsg().getMessageID(), msg.getTopic(), handler.getID()); - } + LOG.debug("Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " + + "interceptorId = {}.", + msg.getMsg().getClientID(), msg.getMsg().getMessageID(), msg.getTopic(), handler.getID()); executor.execute(new Runnable() { @Override public void run() { diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java index c2c72390..1ea90f40 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -29,10 +29,8 @@ class InternalRepublisher { //fire as retained the message Integer packetID = storedMsg.getQos() == AbstractMessage.QOSType.MOST_ONE ? null : targetSession.nextPacketId(); if (packetID != null) { - if (LOG.isDebugEnabled()) { - LOG.debug("Adding message to inflight zone. MqttClientId = {}, packetId = {}, messageId = {}, topic = {}.", - targetSession.clientID, packetID, storedMsg.getMessageID(), storedMsg.getTopic()); - } + LOG.debug("Adding message to inflight zone. MqttClientId = {}, packetId = {}, messageId = {}, topic = {}.", + targetSession.clientID, packetID, storedMsg.getMessageID(), storedMsg.getTopic()); targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID); } PublishMessage publishMsg = retainedPublish(storedMsg); @@ -50,10 +48,8 @@ class InternalRepublisher { for (IMessagesStore.StoredMessage pubEvt : storedPublishes) { //put in flight zone - if (LOG.isDebugEnabled()) { - LOG.debug("Adding message ot inflight zone. MqttClientId = {}, guid = {}, messageId = {}, topic = {}.", - clientSession.clientID, pubEvt.getGuid(), pubEvt.getMessageID(), pubEvt.getTopic()); - } + LOG.debug("Adding message ot inflight zone. MqttClientId = {}, guid = {}, messageId = {}, topic = {}.", + clientSession.clientID, pubEvt.getGuid(), pubEvt.getMessageID(), pubEvt.getTopic()); clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID()); PublishMessage publishMsg = notRetainedPublish(pubEvt); //set the PacketIdentifier only for QoS > 0 diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 5be23eee..07a663e7 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -61,11 +61,8 @@ class MessagesPublisher { ByteBuffer message = origMessage.duplicate(); if (targetIsActive) { - if (LOG.isDebugEnabled()) { - LOG.debug( - "Sending PUBLISH message to subscriber. MessageId = {}, mqttClientId = {}, topicFilter = {}, qos = {}, active = {}.", - pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); - } + LOG.debug("Sending PUBLISH message to active subscriber. MessageId = {}, mqttClientId = {}, topicFilter = {}, " + + "qos = {}.", pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos); PublishMessage publishMsg = notRetainedPublish(topic, qos, message); if (qos != AbstractMessage.QOSType.MOST_ONE) { //QoS 1 or 2 @@ -77,11 +74,9 @@ class MessagesPublisher { this.messageSender.sendPublish(targetSession, publishMsg); } else { if (!targetSession.isCleanSession()) { - if (LOG.isDebugEnabled()) { - LOG.debug( - "Storing pending PUBLISH message. MessageId = {}, mqttClientId = {}, topicFilter = {}, qos = {}, active = {}.", - pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); - } + LOG.debug("Storing pending PUBLISH inactive message. MessageId = {}, mqttClientId = {}, " + + "topicFilter = {}, qos = {}.", + pubMsg.getMessageID(), sub.getClientId(), sub.getTopicFilter(), qos); //store the message in targetSession queue to deliver targetSession.enqueue(pubMsg); } 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 69eec972..140bbb38 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -577,9 +577,7 @@ public class ProtocolProcessor { int messageID = msg.getMessageID(); targetSession.moveInFlightToSecondPhaseAckWaiting(messageID); //once received a PUBREC reply with a PUBREL(messageID) - if (LOG.isDebugEnabled()) { - LOG.debug("Processing PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID); - } + LOG.debug("Processing PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID); PubRelMessage pubRelMessage = new PubRelMessage(); pubRelMessage.setMessageID(messageID); pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE); @@ -590,9 +588,7 @@ public class ProtocolProcessor { public void processPubComp(Channel channel, PubCompMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); - if (LOG.isDebugEnabled()) { - LOG.debug("Processing PUBCOMP message. MqttClientId = {}, messageId = {}.", clientID, messageID); - } + LOG.debug("Processing PUBCOMP message. MqttClientId = {}, messageId = {}.", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID); @@ -733,9 +729,7 @@ public class ProtocolProcessor { return; } - if (LOG.isDebugEnabled()) { - LOG.debug("Removing subscription. MqttClientId = {}, topic = {}.", clientID, topic); - } + LOG.debug("Removing subscription. MqttClientId = {}, topic = {}.", clientID, topic); subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java index 702a6d7e..5c580b31 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -45,8 +45,8 @@ class Qos0PublishHandler extends QosPublishHandler { String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - if (LOG.isDebugEnabled()) { - LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", + if (LOG.isTraceEnabled()) { + LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()), subscriptions.dumpTree()); } else { diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 3b9e5251..afef834b 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -50,8 +50,8 @@ class Qos1PublishHandler extends QosPublishHandler { String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - if (LOG.isDebugEnabled()) { - LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", + if (LOG.isTraceEnabled()) { + LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()), subscriptions.dumpTree()); } else { diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index fe0c59ed..c0ee0b5a 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -55,8 +55,8 @@ class Qos2PublishHandler extends QosPublishHandler { String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - if (LOG.isDebugEnabled()) { - LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", + if (LOG.isTraceEnabled()) { + LOG.trace("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()), subscriptions.dumpTree()); } else { @@ -119,18 +119,14 @@ class Qos2PublishHandler extends QosPublishHandler { private void sendPubRec(String clientID, int messageID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Sending PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID); - } + LOG.debug("Sending PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); connectionDescriptors.sendMessage(pubRecMessage,messageID, clientID); } private void sendPubComp(String clientID, int messageID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Sending PUBCOMP message. MqttClientId = {}, messageId = {}.", clientID, messageID); - } + LOG.debug("Sending PUBCOMP message. MqttClientId = {}, messageId = {}.", clientID, messageID); PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); connectionDescriptors.sendMessage(pubCompMessage, messageID, clientID); diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 0d574db5..61f87875 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -108,16 +108,16 @@ public class SubscriptionsStore { m_sessionsStore = sessionsStore; List subscriptions = sessionsStore.listAllSubscriptions(); //reload any subscriptions persisted - if (LOG.isDebugEnabled()) { - LOG.debug("Reloading all stored subscriptions. SubscriptionTree = {}.", dumpTree()); + if (LOG.isTraceEnabled()) { + LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}.", dumpTree()); } for (ClientTopicCouple clientTopic : subscriptions) { LOG.info("Re-subscribing client to topic. ClientId = {}, topicFilter = {}.", clientTopic.clientID, clientTopic.topicFilter); add(clientTopic); } - if (LOG.isDebugEnabled()) { - LOG.debug("The stored subscriptions have been reloaded. SubscriptionTree = {}.", dumpTree()); + if (LOG.isTraceEnabled()) { + LOG.trace("The stored subscriptions have been reloaded. SubscriptionTree = {}.", dumpTree()); } } @@ -133,14 +133,12 @@ public class SubscriptionsStore { couple.createdNode.addSubscription(newSubscription); //createdNode could be null? //spin lock repeating till we can, swap root, if can't swap just re-do the operation } while(!subscriptions.compareAndSet(oldRoot, couple.root)); - if (LOG.isDebugEnabled()) { - LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot); - } + LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot); } protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) { - List tokens = new ArrayList<>(); + List tokens; try { tokens = parseTopic(topic); } catch (ParseException ex) { 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 b4418afc..172e0a2c 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -55,18 +55,13 @@ class MapDBMessagesStore implements IMessagesStore { @Override public void storeRetained(String topic, MessageGUID guid) { - if (LOG.isDebugEnabled()) { - LOG.debug("Storing retained messages. Topic = {}, guid = {}.", topic, guid); - } + LOG.debug("Storing retained messages. Topic = {}, guid = {}.", topic, guid); m_retainedStore.put(topic, guid); } @Override public Collection searchMatching(IMatchingCondition condition) { - if (LOG.isDebugEnabled()) { - LOG.debug("Scanning retained messages..."); - } - + LOG.debug("Scanning retained messages..."); List results = new ArrayList<>(); for (Map.Entry entry : m_retainedStore.entrySet()) { final MessageGUID guid = entry.getValue(); @@ -76,8 +71,8 @@ class MapDBMessagesStore implements IMessagesStore { } } - if (LOG.isDebugEnabled()) { - LOG.debug("The retained messages have been scanned. MatchingMessages = {}.", results); + if (LOG.isTraceEnabled()) { + LOG.trace("The retained messages have been scanned. MatchingMessages = {}.", results); } return results; @@ -88,10 +83,8 @@ class MapDBMessagesStore implements IMessagesStore { 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); - if (LOG.isDebugEnabled()) { - LOG.debug("Storing publish event. MqttClientId = {}, messageId = {}, guid = {}, topic = {}.", storedMessage.getClientID(), - storedMessage.getMessageID(), guid, storedMessage.getTopic()); - } + 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); @@ -100,9 +93,7 @@ class MapDBMessagesStore implements IMessagesStore { @Override public void dropMessagesInSession(String clientID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Dropping stored messages. ClientId = {}.", clientID); - } + LOG.debug("Dropping stored messages. ClientId = {}.", clientID); ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)); for (MessageGUID guid : messageIdToGuid.values()) { removeStoredMessage(guid); @@ -114,27 +105,21 @@ class MapDBMessagesStore implements IMessagesStore { //remove only the not retained and no more referenced StoredMessage storedMessage = m_persistentMessageStore.get(guid); if (!storedMessage.isRetained()) { - if (LOG.isDebugEnabled()) { - LOG.debug("Dropping stored message. ClientId = {}, messageId = {}, guid = {}, topic = {}.", - storedMessage.getClientID(), storedMessage.getMessageID(), guid, storedMessage.getTopic()); - } + 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); - } + 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); - } + 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 e55b2468..a65f1257 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java @@ -102,9 +102,7 @@ public class MapDBPersistentStore { m_scheduler.scheduleWithFixedDelay(new Runnable() { @Override public void run() { - if (LOG.isDebugEnabled()) { - LOG.debug("Committing to MapDB..."); - } + LOG.debug("Committing to MapDB..."); m_db.commit(); } }, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS); 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 6db1f306..8ba213f7 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -71,8 +71,8 @@ class MapDBSessionsStore implements ISessionsStore { final String clientID = newSubscription.getClientId(); m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); - if (LOG.isDebugEnabled()) { - LOG.debug("The subscription has been added. MqttClientId = {}, topics = {}, clientSubscriptions = {}.", + if (LOG.isTraceEnabled()) { + LOG.trace("The subscription has been added. MqttClientId = {}, topics = {}, clientSubscriptions = {}.", newSubscription.getClientId(), newSubscription.getTopicFilter(), m_db.getHashMap("subscriptions_" + clientID)); } @@ -111,8 +111,8 @@ class MapDBSessionsStore implements ISessionsStore { allSubscriptions.add(new ClientTopicCouple(clientID, topicFilter)); } } - if (LOG.isDebugEnabled()) { - LOG.debug("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions); + if (LOG.isTraceEnabled()) { + LOG.trace("The existing subscriptions have been retrieved. Result = {}.", allSubscriptions); } return allSubscriptions; } @@ -133,9 +133,7 @@ class MapDBSessionsStore implements ISessionsStore { ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); subscriptions.addAll(clientSubscriptions.values()); } - if (LOG.isDebugEnabled()) { - LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions); - } + LOG.debug("The existing subscriptions have been retrieved. Result = {}.", subscriptions); return subscriptions; } @@ -213,9 +211,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void inFlightAck(String clientID, int messageID) { - if (LOG.isDebugEnabled()) { - LOG.debug("Acknowledging inflight message. MqttClientId = {}, messageId = {}.", clientID, messageID); - } + LOG.debug("Acknowledging inflight message. MqttClientId = {}, messageId = {}.", clientID, messageID); Map m = this.m_inflightStore.get(clientID); if (m == null) { LOG.warn("Unable to retrieve inflight message record. MqttClientId = {}, messageId = {}.", clientID, messageID); @@ -232,10 +228,7 @@ 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); - } + LOG.debug("Storing inflight message. MqttClientId = {}, messageId = {}, guid = {}.", clientID, messageID, guid); ConcurrentMap m = this.m_inflightStore.get(clientID); if (m == null) { m = new ConcurrentHashMap<>(); @@ -295,15 +288,11 @@ 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); - } + LOG.debug("Mapping message ID to GUID. MqttClientId = {}, messageId = {}.", clientID, messageID); ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); 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); - } + LOG.debug("The message ID has been mapped to a GUID. MqttClientId = {}, messageId = {}, guid = {}.", + clientID, messageID, result); return result; } diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java index 8aa64098..66ae3123 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java @@ -85,9 +85,7 @@ class LoopMQTTHandler extends ChannelInboundHandlerAdapter { long start = System.nanoTime(); LOG.debug("push forward message the topic {}", msg.getTopicName()); - if (LOG.isDebugEnabled()) { - LOG.debug("content <{}>", payload2Str(msg.getPayload())); - } + LOG.debug("content <{}>", payload2Str(msg.getPayload())); String decodedPayload = payload2Str(msg.getPayload()); long sentTime = Long.parseLong(decodedPayload.split("-")[1]); forthNetworkTime.recordValue(start - sentTime); diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java index 12a268fc..25389e29 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java @@ -44,9 +44,7 @@ class NettyPublishReceiverHandler extends ChannelInboundHandlerAdapter { private void handlePublish(MqttPublishMessage msg) { long start = System.nanoTime(); LOG.debug("push forward message the topic {}", msg.variableHeader().topicName()); - if (LOG.isDebugEnabled()) { - LOG.debug("content <{}>", payload2Str(msg.content())); - } + LOG.debug("content <{}>", payload2Str(msg.content())); String decodedPayload = payload2Str(msg.content()); long sentTime = Long.parseLong(decodedPayload.split("-")[1]); forthNetworkTime.recordValue(start - sentTime); diff --git a/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java b/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java index 270328e7..0ee925a4 100644 --- a/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java +++ b/perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java @@ -45,9 +45,7 @@ class PublishReceiverHandler extends ChannelInboundHandlerAdapter { private void handlePublish(PublishMessage msg) { long start = System.nanoTime(); LOG.debug("push forward message the topic {}", msg.getTopicName()); - if (LOG.isDebugEnabled()) { - LOG.debug("content <{}>", payload2Str(msg.getPayload())); - } + LOG.debug("content <{}>", payload2Str(msg.getPayload())); String decodedPayload = payload2Str(msg.getPayload()); long sentTime = Long.parseLong(decodedPayload.split("-")[1]); forthNetworkTime.recordValue(start - sentTime);