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 a478a993..1cda6b42 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 @@ -78,11 +78,11 @@ class ProtocolProcessor implements EventHandler { } void processConnect(ServerChannel session, ConnectMessage msg) { - LOG.info("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); - LOG.info("processConnect sent bad proto ConnAck"); + LOG.warn("processConnect sent bad proto ConnAck"); session.write(badProto); session.close(false); return; @@ -153,8 +153,9 @@ class ProtocolProcessor implements EventHandler { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); - LOG.info("processConnect sent OK ConnAck"); + 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())); if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 @@ -166,10 +167,11 @@ class ProtocolProcessor implements EventHandler { LOG.debug("republishStored invoked"); List publishedEvents = m_storageService.retrivePersistedPublishes(clientID); if (publishedEvents == null) { - LOG.debug("republishStored, no stored publish events"); + LOG.info(String.format("No stored messages for client <%s>", clientID)); return; } + LOG.info(String.format("republishing stored messages to client <%s>", clientID)); for (PublishEvent pubEvt : publishedEvents) { sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID()); @@ -177,7 +179,7 @@ class ProtocolProcessor implements EventHandler { } private void processRemoveAllSubscriptions(String clientID) { - LOG.debug("processRemoveAllSubscriptions invoked"); + LOG.info(String.format("cleaning old saved subscriptions for client <%s>", clientID)); subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 @@ -375,6 +377,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>", clientID)); } /** @@ -397,7 +400,7 @@ class ProtocolProcessor implements EventHandler { void processSubscribe(ServerChannel session, SubscribeMessage msg, String clientID, boolean cleanSession) { - LOG.info(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); + LOG.debug(String.format("processSubscribe invoked from client %s with msgID %d", clientID, msg.getMessageID())); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.values()[req.getQos()]; @@ -413,11 +416,14 @@ class ProtocolProcessor implements EventHandler { for (int i = 0; i < msg.subscriptions().size(); i++) { ackMessage.addType(AbstractMessage.QOSType.MOST_ONE); } - LOG.info("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", + newSubscription.getClientId(), topic, + AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()))); subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index 4fcd9953..78e230dd 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -25,6 +25,7 @@ public class Server { final Server server = new Server(); server.startServer(); + System.out.println("Server started, version 0.4-SNAPSHOT"); //Bind a shutdown hook Runtime.getRuntime().addShutdownHook(new Thread() { @Override diff --git a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java index 61e1c332..0f9b0ac7 100644 --- a/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java +++ b/parser_commons/src/main/java/org/dna/mqtt/moquette/proto/messages/AbstractMessage.java @@ -24,6 +24,10 @@ public abstract class AbstractMessage { public static enum QOSType { MOST_ONE, LEAST_ONE, EXACTLY_ONCE, RESERVED; + + public static String formatQoS(QOSType qos) { + return String.format("%d - %s", qos.ordinal(), qos.name()); + } } //type protected boolean m_dupFlag;