diff --git a/broker/pom.xml b/broker/pom.xml index 0bb73b19..1b0ee69d 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -41,7 +41,7 @@ org.slf4j slf4j-log4j12 - 1.6.4 + 1.7.5 diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index af7f7a1d..f0349be8 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -146,7 +146,7 @@ public class HawtDBStorageService implements IStorageService { } public Collection searchMatching(IMatchingCondition condition) { - LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size()); + LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); List results = new ArrayList(); @@ -171,7 +171,7 @@ public class HawtDBStorageService implements IStorageService { storedEvents.add(convertToStored(evt)); m_persistentMessageStore.put(clientID, storedEvents); //NB rewind the evt message content - LOG.debug(String.format("Stored published message for client <%s> on topic <%s>", clientID, evt.getTopic())); + LOG.debug("Stored published message for client <{}> on topic <{}>", clientID, evt.getTopic()); } public List retrivePersistedPublishes(String clientID) { @@ -233,7 +233,7 @@ public class HawtDBStorageService implements IStorageService { /*-------- QoS 2 storage management --------------*/ public void persistQoS2Message(String publishKey, PublishEvent evt) { - LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt)); + LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt); m_qos2Store.put(publishKey, convertToStored(evt)); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java index c4625669..7570832e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java @@ -46,7 +46,7 @@ public class MemoryStorageService implements IStorageService { } public Collection searchMatching(IMatchingCondition condition) { - LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size()); + LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); List results = new ArrayList(); @@ -61,7 +61,7 @@ public class MemoryStorageService implements IStorageService { } public void storePublishForFuture(PublishEvent evt) { - LOG.debug("storePublishForFuture store evt " + evt); + LOG.debug("storePublishForFuture store evt {}", evt); List storedEvents; String clientID = evt.getClientID(); if (!m_persistentMessageStore.containsKey(clientID)) { @@ -118,7 +118,7 @@ public class MemoryStorageService implements IStorageService { } public void persistQoS2Message(String publishKey, PublishEvent evt) { - LOG.debug(String.format("persistQoS2Message store pubKey %s, evt %s", publishKey, evt)); + LOG.debug("persistQoS2Message store pubKey {}, evt {}", publishKey, evt); m_qos2Store.put(publishKey, evt); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 027db096..ae1eb3ad 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -79,7 +79,7 @@ class ProtocolProcessor implements EventHandler { } void processConnect(ServerChannel session, ConnectMessage msg) { - LOG.debug("processConnect for client " + msg.getClientID()); + LOG.debug("processConnect for client {}", msg.getClientID()); if (msg.getProcotolVersion() != 0x03) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); @@ -113,7 +113,7 @@ class ProtocolProcessor implements EventHandler { m_clientIDs.put(msg.getClientID(), connDescr); int keepAlive = msg.getKeepAlive(); - LOG.debug(String.format("Connect with keepAlive %d s", keepAlive)); + LOG.debug("Connect with keepAlive {} s", keepAlive); session.setAttribute(Constants.KEEP_ALIVE, keepAlive); session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession()); //used to track the client in the subscription and publishing phases. @@ -158,7 +158,7 @@ class ProtocolProcessor implements EventHandler { okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); LOG.debug("processConnect sent OK ConnAck"); session.write(okResp); - LOG.info(String.format("Connected client ID <%s> with clean session %s", msg.getClientID(), msg.isCleanSession())); + LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 @@ -170,11 +170,11 @@ class ProtocolProcessor implements EventHandler { LOG.debug("republishStored invoked"); List publishedEvents = m_storageService.retrivePersistedPublishes(clientID); if (publishedEvents == null) { - LOG.info(String.format("No stored messages for client <%s>", clientID)); + LOG.info("No stored messages for client <{}>", clientID); return; } - LOG.info(String.format("republishing stored messages to client <%s>", clientID)); + LOG.info("republishing stored messages to client <{}>", clientID); for (PublishEvent pubEvt : publishedEvents) { sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID()); @@ -182,7 +182,7 @@ class ProtocolProcessor implements EventHandler { } private void processRemoveAllSubscriptions(String clientID) { - LOG.info(String.format("cleaning old saved subscriptions for client <%s>", clientID)); + LOG.info("cleaning old saved subscriptions for client <{}>", clientID); subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 @@ -196,8 +196,8 @@ class ProtocolProcessor implements EventHandler { final ByteBuffer message = evt.getMessage(); boolean retain = evt.isRetain(); - LOG.info(String.format("Publish recieved from clientID <%s> on topic <%s> with QoS %s", - evt.getClientID(), evt.getTopic(), evt.getQos())); + LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}", + evt.getClientID(), evt.getTopic(), evt.getQos()); String publishKey = null; if (qos == AbstractMessage.QOSType.LEAST_ONE) { @@ -219,7 +219,7 @@ class ProtocolProcessor implements EventHandler { } m_storageService.cleanInFlight(publishKey); sendPubAck(new PubAckEvent(evt.getMessageID(), evt.getClientID())); - LOG.debug("replying with PubAck to MSG ID " + evt.getMessageID()); + LOG.debug("replying with PubAck to MSG ID {}", evt.getMessageID()); } if (retain) { @@ -231,10 +231,10 @@ class ProtocolProcessor implements EventHandler { * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 * */ private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retain, Integer messageID) { - LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic " + topic); + LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic); for (final Subscription sub : subscriptions.matches(topic)) { - LOG.debug(String.format("Broker publishing to client <%s> topic <%s> qos <%s>", - sub.getClientId(), sub.getTopic(), qos)); + LOG.debug("Broker publishing to client <{}> topic <{}> qos <{}>", + sub.getClientId(), sub.getTopic(), qos); if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS 0 sendPublish(sub.getClientId(), topic, qos, message, false); @@ -271,7 +271,7 @@ class ProtocolProcessor implements EventHandler { pubMessage.setQos(qos); pubMessage.setPayload(message); - LOG.info(String.format("send publish message to <%s> on topic <%s>", clientId, topic)); + LOG.info("send publish message to <{}> on topic <{}>", clientId, topic); if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { pubMessage.setMessageID(messageID); } @@ -280,11 +280,11 @@ class ProtocolProcessor implements EventHandler { if (m_clientIDs == null) { throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); } - LOG.debug("clientIDs are " + m_clientIDs); + LOG.debug("clientIDs are {}", m_clientIDs); if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs)); } - LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); + LOG.debug("Session for clientId {} is {}", clientId, m_clientIDs.get(clientId).getSession()); // m_clientIDs.get(clientId).getSession().write(pubMessage); disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubMessage)); }catch(Throwable t) { @@ -293,7 +293,7 @@ class ProtocolProcessor implements EventHandler { } private void sendPubRec(String clientID, int messageID) { - LOG.debug(String.format("sendPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + LOG.debug("sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); @@ -313,11 +313,11 @@ class ProtocolProcessor implements EventHandler { if (m_clientIDs == null) { throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); } - LOG.debug("clientIDs are " + m_clientIDs); - if (m_clientIDs.get(clientId) == null) { + LOG.debug("clientIDs are {}", m_clientIDs); + if (m_clientIDs.get(clientId) == null) { throw new RuntimeException(String.format("Can't find a ConnectionDEwcriptor for client %s in cache %s", clientId, m_clientIDs)); } - LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); +// LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession()); // m_clientIDs.get(clientId).getSession().write(pubAckMessage); disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientId).getSession(), pubAckMessage)); }catch(Throwable t) { @@ -348,7 +348,7 @@ class ProtocolProcessor implements EventHandler { } private void sendPubComp(String clientID, int messageID) { - LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); + LOG.debug("sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID); PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); @@ -358,7 +358,7 @@ class ProtocolProcessor implements EventHandler { void processPubRec(String clientID, int messageID) { //once received a PUBREC reply with a PUBREL(messageID) - LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + LOG.debug("processPubRec invoked for clientID {} ad messageID {}", clientID, messageID); PubRelMessage pubRelMessage = new PubRelMessage(); pubRelMessage.setMessageID(messageID); @@ -384,7 +384,7 @@ class ProtocolProcessor implements EventHandler { //de-activate the subscriptions for this ClientID // String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID); subscriptions.deactivate(clientID); - LOG.info(String.format("Disconnected client <%s> with clean session %s", clientID, cleanSession)); + LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession); } void proccessConnectionLost(String clientID) { @@ -393,7 +393,7 @@ class ProtocolProcessor implements EventHandler { //de-activate the subscriptions for this ClientID subscriptions.deactivate(clientID); - LOG.info(String.format("Lost connection with client <%s>", clientID)); + LOG.info("Lost connection with client <{}>", clientID); } } @@ -411,13 +411,13 @@ class ProtocolProcessor implements EventHandler { UnsubAckMessage ackMessage = new UnsubAckMessage(); ackMessage.setMessageID(messageID); - LOG.info(String.format("replying with UnsubAck to MSG ID %s", messageID)); + LOG.info("replying with UnsubAck to MSG ID {}", messageID); session.write(ackMessage); } void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { - LOG.debug(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); + LOG.debug("processSubscribe invoked from client {} with msgID {}", clientID, msg.getMessageID()); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; @@ -433,14 +433,14 @@ class ProtocolProcessor implements EventHandler { for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } - LOG.debug("replying with SubAck to MSG ID " + msg.getMessageID()); + LOG.debug("replying with SubAck to MSG ID {}", msg.getMessageID()); session.write(ackMessage); } private void subscribeSingleTopic(Subscription newSubscription, final String topic) { - LOG.info(String.format("<%s> subscribed to topic <%s> with QoS %s", + LOG.info("<{}> subscribed to topic <{}> with QoS {}", newSubscription.getClientId(), topic, - AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()))); + AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())); subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription @@ -452,13 +452,13 @@ class ProtocolProcessor implements EventHandler { for (HawtDBStorageService.StoredMessage storedMsg : messages) { //fire the as retained the message - LOG.debug("send publish message for topic " + topic); + LOG.debug("send publish message for topic {}", topic); sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); } } private void disruptorPublish(OutputMessagingEvent msgEvent) { - LOG.debug("disruptorPublish publishing event on output " + msgEvent); + LOG.debug("disruptorPublish publishing event on output {}", msgEvent); long sequence = m_ringBuffer.next(); ValueEvent event = m_ringBuffer.get(sequence); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index c9238085..681c7044 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -69,7 +69,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { private void disruptorPublish(MessagingEvent msgEvent) { - LOG.debug("disruptorPublish publishing event " + msgEvent); + LOG.debug("disruptorPublish publishing event {}", msgEvent); long sequence = m_ringBuffer.next(); ValueEvent event = m_ringBuffer.get(sequence); @@ -97,7 +97,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { public void onEvent(ValueEvent t, long l, boolean bln) throws Exception { MessagingEvent evt = t.getEvent(); - LOG.info("onEvent processing messaging event from input ringbuffer" + evt); + LOG.info("onEvent processing messaging event from input ringbuffer {}", evt); if (evt instanceof PublishEvent) { m_processor.processPublish((PublishEvent) evt); } else if (evt instanceof StopEvent) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java index 3871d843..9708460e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java @@ -72,7 +72,7 @@ public class SubscriptionsStore { //reload any subscriptions persisted LOG.debug("Reloading all stored subscriptions..."); for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) { - LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic()); + LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopic()); addDirect(subscription); } LOG.debug("Finished loading"); @@ -166,7 +166,7 @@ public class SubscriptionsStore { } public void activate(String clientID) { - LOG.debug(String.format("Activating subscriptions for clientID <%s>", clientID)); + LOG.debug("Activating subscriptions for clientID <{}>", clientID); subscriptions.activate(clientID); } @@ -176,7 +176,7 @@ public class SubscriptionsStore { * listeners subscriptions, and not topic publishing. */ public List matches(String topic) { - List tokens = new ArrayList(); + List tokens; try { tokens = splitTopic(topic); } catch (ParseException ex) { diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java index 22c65a2c..da0cf1cd 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java @@ -85,7 +85,7 @@ public class NettyAcceptor implements ServerAcceptor { MessageMetrics metrics = m_metricsCollector.computeMetrics(); //LOG.info(String.format("Bytes read: %d, bytes wrote: %d", metrics.readBytes(), metrics.wroteBytes())); - LOG.info(String.format("Msg read: %d, msg wrote: %d", metrics.messagesRead(), metrics.messagesWrote())); + LOG.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote()); } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java index 88b8dff5..4e6d1aa0 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java @@ -28,7 +28,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object message) { AbstractMessage msg = (AbstractMessage) message; - LOG.info(String.format("Received a message of type %s", Utils.msgType2String(msg.getMessageType()))); + LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType())); try { switch (msg.getMessageType()) { case CONNECT: diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java index 593d8a5b..bdace02b 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/PublishDecoder.java @@ -19,14 +19,14 @@ class PublishDecoder extends DemuxDecoder { @Override void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { - LOG.info("decode invoked with buffer " + in); + LOG.info("decode invoked with buffer {}", in); in.resetReaderIndex(); int startPos = in.readerIndex(); //Common decoding part PublishMessage message = new PublishMessage(); if (!decodeCommonHeader(message, in)) { - LOG.info("decode ask for more data after " + in); + LOG.info("decode ask for more data after {}", in); in.resetReaderIndex(); return; }