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

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

Browse Source
This commit is contained in:
andsel
2017-01-18 23:28:11 +01:00
parent e295fcf9f2
commit 716c9ac0a0
17 changed files with 72 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/Server.java
+4 -6
View File
@@ -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);
}
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+1 -3
View File
@@ -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:
broker/src/main/java/io/moquette/spi/ClientSession.java
+1 -1
View File
@@ -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;
broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java
+15 -28
View File
@@ -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() {
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+4 -8
View File
@@ -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
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+5 -10
View File
@@ -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);
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+3 -9
View File
@@ -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);
broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java
+2 -2
View File
@@ -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 {
broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java
+2 -2
View File
@@ -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 {
broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java
+4 -8
View File
@@ -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);
broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+6 -8
View File
@@ -108,16 +108,16 @@ public class SubscriptionsStore {
m_sessionsStore = sessionsStore;
List<ClientTopicCouple> 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<Token> tokens = new ArrayList<>();
List<Token> tokens;
try {
tokens = parseTopic(topic);
} catch (ParseException ex) {
broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+11 -26
View File
@@ -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<StoredMessage> searchMatching(IMatchingCondition condition) {
if (LOG.isDebugEnabled()) {
LOG.debug("Scanning retained messages...");
}
LOG.debug("Scanning retained messages...");
List<StoredMessage> results = new ArrayList<>();
for (Map.Entry<String, MessageGUID> 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<Integer, MessageGUID> 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<Integer, MessageGUID> 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);
}
broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java
+1 -3
View File
@@ -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);
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+10 -21
View File
@@ -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<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);
}
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<Integer, MessageGUID> 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<Integer, MessageGUID> 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<Integer, MessageGUID> 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;
}
perf/src/main/java/io/moquette/parser/netty/performance/LoopMQTTHandler.java
+1 -3
View File
@@ -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);
perf/src/main/java/io/moquette/parser/netty/performance/NettyPublishReceiverHandler.java
+1 -3
View File
@@ -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);
perf/src/main/java/io/moquette/parser/netty/performance/PublishReceiverHandler.java
+1 -3
View File
@@ -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);