From 8b3f158a436fcbbcf2e6ed9dc422120a23a01929 Mon Sep 17 00:00:00 2001 From: Luis Barrios Date: Wed, 11 Jan 2017 18:42:30 +0100 Subject: [PATCH] Improve logging. Remove duplicated code from QosPublishHandlers. --- .../spi/impl/InternalRepublisher.java | 10 +- .../moquette/spi/impl/MessagesPublisher.java | 15 +- .../moquette/spi/impl/ProtocolProcessor.java | 164 ++++++++++++------ .../impl/ProtocolProcessorBootstrapper.java | 140 ++++++++------- .../moquette/spi/impl/Qos0PublishHandler.java | 29 ++-- .../moquette/spi/impl/Qos1PublishHandler.java | 32 ++-- .../moquette/spi/impl/Qos2PublishHandler.java | 55 +++--- .../moquette/spi/impl/QosPublishHandler.java | 30 ++++ 8 files changed, 285 insertions(+), 190 deletions(-) create mode 100644 broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java 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 b568dc2b..c2c72390 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -29,7 +29,10 @@ class InternalRepublisher { //fire as retained the message Integer packetID = storedMsg.getQos() == AbstractMessage.QOSType.MOST_ONE ? null : targetSession.nextPacketId(); if (packetID != null) { - LOG.trace("Adding to inflight <{}>", packetID); + if (LOG.isDebugEnabled()) { + 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); @@ -47,7 +50,10 @@ class InternalRepublisher { for (IMessagesStore.StoredMessage pubEvt : storedPublishes) { //put in flight zone - LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID()); + if (LOG.isDebugEnabled()) { + 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 95988f6c..5be23eee 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -52,18 +52,20 @@ class MessagesPublisher { if (publishingQos != AbstractMessage.QOSType.MOST_ONE) { guid = m_messagesStore.storePublishForFuture(pubMsg); } - - LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic); + for (final Subscription sub : topicMatchingSubscriptions) { AbstractMessage.QOSType qos = lowerQosToTheSubscriptionDesired(sub, publishingQos); ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId()); - LOG.debug("Broker republishing to client <{}> topicFilter <{}> qos <{}>, active {}", - sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); 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); + } PublishMessage publishMsg = notRetainedPublish(topic, qos, message); if (qos != AbstractMessage.QOSType.MOST_ONE) { //QoS 1 or 2 @@ -75,6 +77,11 @@ 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); + } //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 a8913ee4..3767f6da 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -185,6 +185,7 @@ public class ProtocolProcessor { IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { + LOG.info("Initializing MQTT protocol processor..."); this.connectionDescriptors = connectionDescriptors; this.subscriptionInCourse = new ConcurrentHashMap<>(); this.m_interceptor = interceptor; @@ -192,15 +193,19 @@ public class ProtocolProcessor { this.allowAnonymous = allowAnonymous; this.allowZeroByteClientId = allowZeroByteClientId; m_authorizator = authorizator; - LOG.trace("subscription tree on init {}", subscriptions.dumpTree()); + if (LOG.isDebugEnabled()) { + LOG.debug("Initial subscriptions tree = {}.", subscriptions.dumpTree()); + } m_authenticator = authenticator; m_messagesStore = storageService; m_sessionsStore = sessionsStore; m_server_port = serverPort; + LOG.info("Initializing messages publisher..."); final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors); this.messagesPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, m_messagesStore, messageSender); + LOG.info("Initializing QoS publish handlers..."); this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor, this.messagesPublisher); this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, subscriptions, m_messagesStore, @@ -208,16 +213,19 @@ public class ProtocolProcessor { this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port, this.messagesPublisher); + LOG.info("Initializing internal republisher..."); this.internalRepublisher = new InternalRepublisher(messageSender); } public void processConnect(Channel channel, ConnectMessage msg) { - LOG.info("CONNECT for client <{}>", msg.getClientID()); + LOG.info("Processing CONNECT message. MqttClientId = {}, username = {}, password = {}.", msg.getClientID(), + msg.getUsername(), msg.getPassword()); if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); - LOG.warn("CONNECT sent bad proto ConnAck"); + LOG.error("The MQTT protocol version is not valid. MqttClientId = {}, protocolVersion = {}.", + msg.getProtocolVersion()); channel.writeAndFlush(badProto); channel.close(); return; @@ -229,14 +237,16 @@ public class ProtocolProcessor { okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); channel.writeAndFlush(okResp); channel.close(); - LOG.warn("CONNECT sent rejected identifier ConnAck"); + LOG.error("The MQTT client ID cannot be empty., Username = {}, password = {}.", msg.getUsername(), + msg.getPassword()); return; } // Generating client id. String randomIdentifier = UUID.randomUUID().toString().replace("-", ""); msg.setClientID(randomIdentifier); - LOG.info("Client connected with server generated identifier: {}", randomIdentifier); + LOG.info("The client has connected with a server generated identifier. MqttClientId = {}, username = {}, password = {}.", + randomIdentifier, msg.getUsername(), msg.getPassword()); } if (!login(channel, msg)) { @@ -248,7 +258,8 @@ public class ProtocolProcessor { ConnectionDescriptor descriptor = new ConnectionDescriptor(clientID, channel, msg.isCleanSession()); ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor); if (existing != null) { - LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); + LOG.info("The client ID is being used in an existing connection. It will be closed. MqttClientId = {}.", + msg.getClientID()); existing.abort(); return; } @@ -277,10 +288,10 @@ public class ProtocolProcessor { final boolean success = descriptor.assignState(ConnectionState.MESSAGES_REPUBLISHED, ConnectionState.ESTABLISHED); if (!success) { channel.close(); - } else { - LOG.info("Connection established"); } - LOG.info("CONNECT processed"); + + LOG.info("The CONNECT message has been processed. MqttClientId = {}, username = {}, password = {}.", + msg.getClientID(), msg.getUsername(), msg.getPassword()); } private boolean login(Channel channel, ConnectMessage msg) { @@ -290,15 +301,21 @@ public class ProtocolProcessor { if (msg.isPasswordFlag()) { pwd = msg.getPassword(); } else if (!this.allowAnonymous) { + LOG.error("The client didn't supply any password and MQTT anonymous mode is disabled. MqttClientId = {}.", + msg.getClientID()); failedCredentials(channel); return false; } if (!m_authenticator.checkValid(msg.getClientID(), msg.getUsername(), pwd)) { + LOG.error("The authenticator has rejected the MQTT credentials. MqttClientId = {}, username = {}, password = {}.", + msg.getClientID(), msg.getUsername(), pwd); failedCredentials(channel); return false; } NettyUtils.userName(channel, msg.getUsername()); } else if (!this.allowAnonymous) { + LOG.error("The client didn't supply any credentials and MQTT anonymous mode is disabled. MqttClientId = {}.", + msg.getClientID()); failedCredentials(channel); return false; } @@ -306,6 +323,7 @@ public class ProtocolProcessor { } private boolean sendAck(ConnectionDescriptor descriptor, ConnectMessage msg) { + LOG.info("Sending connect ACK. MqttClientId = {}.", msg.getClientID()); final boolean success = descriptor.assignState(ConnectionState.DISCONNECTED, ConnectionState.SENDACK); if (!success) { return false; @@ -319,36 +337,42 @@ public class ProtocolProcessor { okResp.setSessionPresent(true); } if (isSessionAlreadyStored) { + LOG.info("Cleaning session. MqttClientId = {}.", msg.getClientID()); clientSession.cleanSession(msg.isCleanSession()); } descriptor.writeAndFlush(okResp); + LOG.info("The connect ACK has been sent. MqttClientId = {}.", msg.getClientID()); return true; } private void initializeKeepAliveTimeout(Channel channel, ConnectMessage msg) { int keepAlive = msg.getKeepAlive(); - LOG.debug("Connect with keepAlive {} s", keepAlive); + LOG.info("Configuring connection. MqttClientId = {}.", msg.getClientID()); NettyUtils.keepAlive(channel, keepAlive); //session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.isCleanSession()); NettyUtils.cleanSession(channel, msg.isCleanSession()); //used to track the client in the subscription and publishing phases. //session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); NettyUtils.clientID(channel, msg.getClientID()); - LOG.debug("Connect create session <{}>", channel); + int idleTime = Math.round(keepAlive * 1.5f); + setIdleTime(channel.pipeline(), idleTime); - setIdleTime(channel.pipeline(), Math.round(keepAlive * 1.5f)); + LOG.info("The connection has been configured. MqttClientId = {}, keepAlive = {}, cleanSession = {}, idleTime = {}.", + msg.getClientID(), keepAlive, msg.isCleanSession(), idleTime); } private void storeWillMessage(ConnectMessage msg) { //Handle will flag if (msg.isWillFlag()) { AbstractMessage.QOSType willQos = AbstractMessage.QOSType.valueOf(msg.getWillQos()); + LOG.info("Configuring MQTT last will and testament. MqttClientId = {}, willTopic = {}, willQos = {}, willRetain = {}.", + msg.getClientID(), willQos, msg.getWillTopic(), msg.isWillRetain()); byte[] willPayload = msg.getWillMessage(); ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); //save the will testament in the clientID store WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos); m_willStore.put(msg.getClientID(), will); - LOG.info("Session for clientID <{}> with will to topic {}", msg.getClientID(), msg.getWillTopic()); + LOG.info("MQTT last will and testament has been configured. MqttClientId = {}.", msg.getClientID()); } } @@ -361,13 +385,12 @@ public class ProtocolProcessor { ClientSession clientSession = m_sessionsStore.sessionForClient(msg.getClientID()); boolean isSessionAlreadyStored = clientSession != null; if (!isSessionAlreadyStored) { - LOG.debug("Create persistent session for clientID <{}>", msg.getClientID()); clientSession = m_sessionsStore.createNewSession(msg.getClientID(), msg.isCleanSession()); } if (msg.isCleanSession()) { + LOG.info("Cleaning session. MqttClientId = {}.", msg.getClientID()); clientSession.cleanSession(); } - LOG.debug("Created session for client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); return clientSession; } @@ -404,14 +427,13 @@ public class ProtocolProcessor { * Republish QoS1 and QoS2 messages stored into the session for the clientID. * */ private void republishStoredInSession(ClientSession clientSession) { - LOG.trace("republishStoredInSession for client <{}>", clientSession); + LOG.info("Republishing stored publish events. MqttClientId = {}.", clientSession.clientID); BlockingQueue publishedEvents = clientSession.queue(); if (publishedEvents.isEmpty()) { - LOG.info("No stored messages for client <{}>", clientSession.clientID); + LOG.info("There are no stored publish events. ClientId = {}.", clientSession.clientID); return; } - LOG.info("republishing stored messages to client <{}>", clientSession.clientID); this.internalRepublisher.publishStored(clientSession, publishedEvents); } @@ -448,7 +470,8 @@ public class ProtocolProcessor { } public void processPublish(Channel channel, PublishMessage msg) { - LOG.info("PUB --PUBLISH--> SRV executePublish invoked with {}", msg); + LOG.info("Processing PUBLISH message. MqttClientId = {}, topic = {}, messageId = {}, qos = {}.", + msg.getClientId(), msg.getTopicName(), msg.getMessageID(), msg.getQos()); final AbstractMessage.QOSType qos = msg.getQos(); switch (qos) { case MOST_ONE: @@ -478,7 +501,7 @@ public class ProtocolProcessor { public void internalPublish(PublishMessage msg) { final AbstractMessage.QOSType qos = msg.getQos(); final String topic = msg.getTopicName(); - LOG.info("embedded PUBLISH on topic <{}> with QoS {}", topic, qos); + LOG.info("Sending PUBLISH message. Topic = {}, qos = {}.", topic, qos); MessageGUID guid = null; IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); @@ -526,6 +549,7 @@ public class ProtocolProcessor { String topic = tobeStored.getTopic(); List topicMatchingSubscriptions = subscriptions.matches(topic); + LOG.info("Publishing will message. MqttClientId = {}, messageId = {}, topic = {}.", clientID, messageId, topic); this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions); } @@ -553,7 +577,9 @@ public class ProtocolProcessor { int messageID = msg.getMessageID(); targetSession.moveInFlightToSecondPhaseAckWaiting(messageID); //once received a PUBREC reply with a PUBREL(messageID) - LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientID {} ad messageID {}", clientID, messageID); + if (LOG.isDebugEnabled()) { + LOG.debug("Processing PUBREC message. MqttClientId = {}, messageId = {}.", clientID, messageID); + } PubRelMessage pubRelMessage = new PubRelMessage(); pubRelMessage.setMessageID(messageID); pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE); @@ -564,7 +590,9 @@ public class ProtocolProcessor { public void processPubComp(Channel channel, PubCompMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); - LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); + if (LOG.isDebugEnabled()) { + 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); @@ -574,8 +602,9 @@ public class ProtocolProcessor { } public void processDisconnect(Channel channel) throws InterruptedException { - channel.flush(); - final String clientID = NettyUtils.clientID(channel); + final String clientID = NettyUtils.clientID(channel); + LOG.info("Processing DISCONNECT message. MqttClientId = {}.", clientID); + channel.flush(); final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID); if (existingDescriptor == null) { //another client with same ID removed the descriptor, we must exit @@ -585,36 +614,43 @@ public class ProtocolProcessor { if (existingDescriptor.doesNotUseChannel(channel)) { //another client saved it's descriptor, exit - existingDescriptor.abort(); + LOG.warn("Another client is using the connection descriptor. Closing connection. MqttClientId = {}.", + clientID); + existingDescriptor.abort(); return; } if (!removeSubscriptions(existingDescriptor, clientID)) { - existingDescriptor.abort(); + LOG.warn("Unable to remove subscriptions. Closing connection. MqttClientId = {}.", clientID); + existingDescriptor.abort(); return; } if (!dropStoredMessages(existingDescriptor, clientID)) { - existingDescriptor.abort(); + LOG.warn("Unable to drop stored messages. Closing connection. MqttClientId = {}.", clientID); + existingDescriptor.abort(); return; } if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) { - existingDescriptor.abort(); + LOG.warn("Unable to drop will message. Closing connection. MqttClientId = {}.", clientID); + existingDescriptor.abort(); return; } if (!existingDescriptor.close()) { + LOG.info("The connection has been closed. MqttClientId = {}.", clientID); return; } boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor); if (!stillPresent) { //another descriptor was inserted - return; + LOG.warn("Another descriptor has been inserted. MqttClientId = {}.", clientID); + return; } - LOG.info("DISCONNECT client <{}> finished", clientID); + LOG.info("The DISCONNECT message has been processed. MqttClientId = {}.", clientID); } private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) { @@ -624,9 +660,9 @@ public class ProtocolProcessor { } if (descriptor.cleanSession) { - LOG.info("cleaning old saved subscriptions for client <{}>", clientID); + LOG.info("Removing saved subscriptions. MqttClientId = {}.", descriptor.clientID); m_sessionsStore.wipeSubscriptions(clientID); - LOG.debug("Wiped subscriptions for client <{}>", clientID); + LOG.info("The saved subscriptions have been removed. MqttClientId = {}.", descriptor.clientID); } return true; } @@ -638,9 +674,9 @@ public class ProtocolProcessor { } if (descriptor.cleanSession) { - LOG.debug("Removing messages in session's queue for client <{}>", clientID); + LOG.debug("Removing messages of session. MqttClientId = {}.", descriptor.clientID); this.m_sessionsStore.dropQueue(clientID); - LOG.debug("Removed messages in session for client's queue <{}>", clientID); + LOG.debug("The messages of the session have been removed. MqttClientId = {}.", descriptor.clientID); } return true; } @@ -651,6 +687,7 @@ public class ProtocolProcessor { return false; } + LOG.info("Removing will message. ClientId = {}.", descriptor.clientID); //cleanup the will store m_willStore.remove(clientID); String username = descriptor.getUsername(); @@ -659,6 +696,7 @@ public class ProtocolProcessor { } public void processConnectionLost(String clientID, Channel channel) { + LOG.info("Processing connection lost event. MqttClientId = {}.", clientID); ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true); connectionDescriptors.removeConnection(oldConnDescr); //publish the Will message (if any) for the clientID @@ -682,7 +720,7 @@ public class ProtocolProcessor { List topics = msg.topicFilters(); String clientID = NettyUtils.clientID(channel); - LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); + LOG.info("Processing UNSUBSCRIBE message. MqttClientId = {}, topics = {}.", clientID, topics); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); for (String topic : topics) { @@ -690,10 +728,14 @@ public class ProtocolProcessor { if (!validTopic) { //close the connection, not valid topicFilter is a protocol violation channel.close(); - LOG.warn("UNSUBSCRIBE found an invalid topic filter <{}> for clientID <{}>", topic, clientID); + LOG.error("The topic filter is not valid. MqttClientId = {}, topics = {}, badTopicFilter = {}.", + clientID, topics, topic); return; } + if (LOG.isDebugEnabled()) { + LOG.debug("Removing subscription. MqttClientId = {}, topic = {}.", clientID, topic); + } subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); @@ -705,43 +747,45 @@ public class ProtocolProcessor { UnsubAckMessage ackMessage = new UnsubAckMessage(); ackMessage.setMessageID(messageID); - LOG.info("replying with UnsubAck to MSG ID {}", messageID); + LOG.info("Sending UNSUBACK message. MqttClientId = {}, topics = {}, messageId = {}.", clientID, topics, + messageID); channel.writeAndFlush(ackMessage); } public void processSubscribe(Channel channel, SubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); - LOG.info("SUBSCRIBE client <{}>", clientID); int messageID = msg.getMessageID(); - LOG.debug("SUBSCRIBE client <{}> on server {} packetID {}", clientID, m_server_port, messageID); + LOG.info("Processing SUBSCRIBE message. MqttClientId = {}, messageId = {}.", clientID, msg.getMessageID()); RunningSubscription executionKey = new RunningSubscription(clientID, messageID); SubscriptionState currentStatus = this.subscriptionInCourse.putIfAbsent(executionKey, SubscriptionState.VERIFIED); if (currentStatus != null) { - LOG.debug("The client <{}> sent another SUBSCRIBE while this one was processing", clientID); + LOG.warn("The client sent another SUBSCRIBE message while this one was being processed. MqttClientId = {}, messageId = {}.", + clientID, msg.getMessageID()); return; } String username = NettyUtils.userName(channel); List ackTopics = doVerify(clientID, username, msg); SubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics); if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { - LOG.debug("The client {} sent another SUBSCRIBE while this one was verifing topicFilters"); + LOG.warn("The client sent another SUBSCRIBE message while the topic filters were being verified. MqttClientId = {}, messageId = {}.", + clientID, msg.getMessageID()); return; } + LOG.info("Creating and storing subscriptions. MqttClientId = {}, messageId = {}, topics = {}.", clientID, + msg.getMessageID(), ackTopics); + ackMessage.setMessageID(messageID); List newSubscriptions = doStoreSubscription(ackTopics, clientID); //save session, persist subscriptions from session - LOG.debug("SUBACK for packetID {}", messageID); - if (LOG.isTraceEnabled()) { - LOG.trace("subscription tree {}", subscriptions.dumpTree()); - } for (Subscription subscription : newSubscriptions) { - LOG.debug("Persisting subscription {}", subscription); subscriptions.add(subscription.asClientTopicCouple()); } + + LOG.info("Sending SUBACK response. MqttClientId = {}, messageId = {}.", clientID, msg.getMessageID()); channel.writeAndFlush(ackMessage); //fire the persisted messages in session @@ -751,7 +795,8 @@ public class ProtocolProcessor { boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); if (!success) { - LOG.warn("Failed to remove the descriptor, something bad happened"); + LOG.warn("Unable to perform the final subscription state update. MqttClientId = {}, messageId = {}.", + clientID, msg.getMessageID()); } } @@ -784,13 +829,20 @@ public class ProtocolProcessor { for (SubscribeMessage.Couple req : msg.subscriptions()) { if (!m_authorizator.canRead(req.topicFilter, username, clientSession.clientID)) { //send SUBACK with 0x80, the user hasn't credentials to read the topic - LOG.debug("topic {} doesn't have read credentials", req.topicFilter); + LOG.error("The client does not have read permissions on the topic. MqttClientId = {}, username = {}, messageId = {}, topic = {}.", + clientID, username, msg.getMessageID(), req.topicFilter); ackTopics.add(new SubscribeMessage.Couple(AbstractMessage.QOSType.FAILURE.byteValue(), req.topicFilter)); } else { - boolean validTopic = SubscriptionsStore.validate(req.topicFilter); - AbstractMessage.QOSType qos = validTopic ? - AbstractMessage.QOSType.valueOf(req.qos) - : AbstractMessage.QOSType.FAILURE; + AbstractMessage.QOSType qos = null; + if (SubscriptionsStore.validate(req.topicFilter)) { + qos = AbstractMessage.QOSType.valueOf(req.qos); + LOG.info("The client will be subscribed to the topic. MqttClientId = {}, username = {}, messageId = {}, topic = {}.", + clientID, username, msg.getMessageID(), req.topicFilter); + } else { + LOG.error("The topic filter is not valid. MqttClientId = {}, username = {}, messageId = {}, topic = {}.", + clientID, username, msg.getMessageID(), req.topicFilter); + qos = AbstractMessage.QOSType.FAILURE; + } ackTopics.add(new SubscribeMessage.Couple(qos.byteValue(), req.topicFilter)); } } @@ -810,7 +862,8 @@ public class ProtocolProcessor { } private void publishRetainedMessagesInSession(final Subscription newSubscription, String username) { - LOG.debug("Publish persisted messages in session {}", newSubscription); + LOG.info("Retrieving retained messages. MqttClientId = {}, topics = {}.", newSubscription.getClientId(), + newSubscription.getTopicFilter()); //scans retained messages to be published to the new subscription //TODO this is ugly, it does a linear scan on potential big dataset @@ -821,7 +874,10 @@ public class ProtocolProcessor { } }); - LOG.debug("Found {} messages to republish", messages.size()); + if (!messages.isEmpty()) { + LOG.info("Publishing retained messages. MqttClientId = {}, topics = {}, messagesNo = {}.", + newSubscription.getClientId(), newSubscription.getTopicFilter(), messages.size()); + } ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId()); this.internalRepublisher.publishRetained(targetSession, messages); diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 13fbb510..cf0358e8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -35,7 +35,6 @@ import io.moquette.spi.security.IAuthorizator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.text.ParseException; @@ -81,36 +80,32 @@ public class ProtocolProcessorBootstrapper { IAuthenticator authenticator, IAuthorizator authorizator, Server server) { subscriptions = new SubscriptionsStore(); + LOG.info("Initializing messages and sessions stores..."); m_mapStorage = new MapDBPersistentStore(props); m_mapStorage.initStore(); IMessagesStore messagesStore = m_mapStorage.messagesStore(); m_sessionsStore = m_mapStorage.sessionsStore(); + LOG.info("Configuring message interceptors..."); + List observers = new ArrayList<>(embeddedObservers); String interceptorClassName = props.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME); if (interceptorClassName != null && !interceptorClassName.isEmpty()) { - try { - InterceptHandler handler; - try { - final Constructor constructor = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).getConstructor(Server.class); - handler = constructor.newInstance(server); - } catch (NoSuchMethodException nsme){ - handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance(); + InterceptHandler handler = loadClass(interceptorClassName, InterceptHandler.class, Server.class, server); + if (handler != null) { + observers.add(handler); } - observers.add(handler); - } catch (Throwable ex) { - LOG.error("Can't load the intercept handler {}", ex); - } } m_interceptor = new BrokerInterceptor(props, observers); + LOG.info("Initializing subscriptions store..."); subscriptions.init(m_sessionsStore); + LOG.info("Configuring MQTT authenticator..."); String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, ""); - if (!authenticatorClassName.isEmpty()) { - authenticator = (IAuthenticator)loadClass(authenticatorClassName, IAuthenticator.class, props); - LOG.info("Loaded custom authenticator {}", authenticatorClassName); + if (authenticator == null && !authenticatorClassName.isEmpty()) { + authenticator = loadClass(authenticatorClassName, IAuthenticator.class, IConfig.class, props); } IResourceLoader resourceLoader = props.getResourceLoader(); @@ -121,12 +116,13 @@ public class ProtocolProcessorBootstrapper { } else { authenticator = new ResourceAuthenticator(resourceLoader, passwdPath); } + LOG.info("An {} authenticator instance will be used.", authenticator.getClass().getName()); } + LOG.info("Configuring MQTT authorizator..."); String authorizatorClassName = props.getProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, ""); - if (!authorizatorClassName.isEmpty()) { - authorizator = (IAuthorizator)loadClass(authorizatorClassName, IAuthorizator.class, props); - LOG.info("Loaded custom authorizator {}", authorizatorClassName); + if (authorizator == null && !authorizatorClassName.isEmpty()) { + authorizator = loadClass(authorizatorClassName, IAuthorizator.class, IConfig.class, props); } if (authorizator == null) { @@ -134,70 +130,82 @@ public class ProtocolProcessorBootstrapper { if (aclFilePath != null && !aclFilePath.isEmpty()) { authorizator = new DenyAllAuthorizator(); try { + LOG.info("Parsing ACL file. Path = {}.", aclFilePath); authorizator = ACLFileParser.parse(resourceLoader.loadResource(aclFilePath)); } catch (ParseException pex) { - LOG.error(String.format("Format error in parsing acl %s %s", resourceLoader.getName(), aclFilePath), pex); + LOG.error("Unable to parse ACL file. Path = {}, cause = {}, errorMessage = {}.", aclFilePath, + pex.getCause(), pex.getMessage()); } - LOG.info("Using acl file defined at path {}", aclFilePath); } else { authorizator = new PermitAllAuthorizator(); - LOG.info("Starting without ACL definition"); } - + LOG.info("An {} authorizator instance will be used.", authorizator.getClass().getName()); } + LOG.info("Initializing MQTT protocol processor..."); boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); boolean allowZeroByteClientId = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false")); m_processor.init(subscriptions, messagesStore, m_sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); return m_processor; } - private Object loadClass(String className, Class cls, IConfig props) { - Object instance = null; - try { - Class clazz = Class.forName(className); + @SuppressWarnings("unchecked") + private T loadClass(String className, Class iface, Class constructorArgClass, U props) { + T instance = null; + try { + LOG.info("Loading class. ClassName = {}, interfaceName = {}.", className, iface.getName()); + Class clazz = Class.forName(className); - // check if method getInstance exists - Method method = clazz.getMethod("getInstance", new Class[] {}); - try { - instance = method.invoke(null, new Object[] {}); - } catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) { - LOG.error(null, ex); - throw new RuntimeException("Cannot call method "+ className +".getInstance", ex); - } - } - catch (NoSuchMethodException nsmex) { - try { - // check if constructor with IConfig parameter exists - instance = this.getClass().getClassLoader() - .loadClass(className) - .asSubclass(cls) - .getConstructor(IConfig.class).newInstance(props); - } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { - LOG.error(null, ex); - throw new RuntimeException("Cannot load custom authenticator class " + className, ex); - } catch (NoSuchMethodException | InvocationTargetException e) { - try { - // fallback to default constructor - instance = this.getClass().getClassLoader() - .loadClass(className) - .asSubclass(cls) - .newInstance(); - } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { - LOG.error(null, ex); - throw new RuntimeException("Cannot load custom authenticator class " + className, ex); - } - } - } catch (ClassNotFoundException ex) { - LOG.error(null, ex); - throw new RuntimeException("Class " + className + " not found", ex); - } catch (SecurityException ex) { - LOG.error(null, ex); - throw new RuntimeException("Cannot call method "+ className +".getInstance", ex); - } + // check if method getInstance exists + Method method = clazz.getMethod("getInstance", new Class[] {}); + try { + LOG.info("Invoking getInstance() method. ClassName = {}, interfaceName = {}.", className, + iface.getName()); + instance = (T) method.invoke(null, new Object[] {}); + } catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) { + LOG.error( + "Unable to invoke getInstance() method. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", + className, iface.getName(), ex.getCause(), ex.getMessage()); + return null; + } + } catch (NoSuchMethodException nsmex) { + try { + // check if constructor with constructor arg class parameter + // exists + LOG.info("Invoking constructor with {} argument. ClassName = {}, interfaceName = {}.", + constructorArgClass.getName(), className, iface.getName()); + instance = this.getClass().getClassLoader().loadClass(className).asSubclass(iface) + .getConstructor(constructorArgClass).newInstance(props); + } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { + LOG.warn( + "Unable to invoke constructor with {} argument. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", + constructorArgClass.getName(), className, iface.getName(), ex.getCause(), ex.getMessage()); + return null; + } catch (NoSuchMethodException | InvocationTargetException e) { + try { + LOG.info("Invoking default constructor. ClassName = {}, interfaceName = {}.", className, + iface.getName()); + // fallback to default constructor + instance = this.getClass().getClassLoader().loadClass(className).asSubclass(iface).newInstance(); + } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) { + LOG.error( + "Unable to invoke default constructor. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", + className, iface.getName(), ex.getCause(), ex.getMessage()); + return null; + } + } + } catch (ClassNotFoundException ex) { + LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, iface.getName()); + return null; + } catch (SecurityException ex) { + LOG.error( + "Unable to load class due to a security violation. ClassName = {}, interfaceName = {}, cause = {}, errorMessage = {}.", + className, iface.getName(), ex.getCause(), ex.getMessage()); + return null; + } - return instance; - } + return instance; + } public List getSubscriptions() { return m_sessionsStore.getSubscriptions(); 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 0adeebd9..702a6d7e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -14,11 +14,10 @@ import java.util.List; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; -class Qos0PublishHandler { +class Qos0PublishHandler extends QosPublishHandler { private static final Logger LOG = LoggerFactory.getLogger(Qos0PublishHandler.class); - private final IAuthorizator m_authorizator; private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; @@ -27,7 +26,7 @@ class Qos0PublishHandler { public Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, MessagesPublisher messagesPublisher) { - this.m_authorizator = authorizator; + super(authorizator); this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; @@ -46,11 +45,15 @@ class Qos0PublishHandler { String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic); - if (LOG.isTraceEnabled()) { - LOG.trace("content <{}>", DebugUtils.payload2Str(toStoreMsg.getMessage())); - LOG.trace("subscription tree {}", subscriptions.dumpTree()); - } + if (LOG.isDebugEnabled()) { + LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", + clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()), + subscriptions.dumpTree()); + } else { + LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID, + topic, msg.getMessageID()); + } + List topicMatchingSubscriptions = subscriptions.matches(topic); this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); @@ -62,14 +65,4 @@ class Qos0PublishHandler { String username = NettyUtils.userName(channel); m_interceptor.notifyTopicPublished(msg, clientID, username); } - - boolean checkWriteOnTopic(String topic, Channel channel) { - String clientID = NettyUtils.clientID(channel); - String username = NettyUtils.userName(channel); - if (!m_authorizator.canWrite(topic, username, clientID)) { - LOG.debug("topic {} doesn't have write credentials", topic); - return true; - } - return false; - } } 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 a458042b..3b9e5251 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -17,27 +17,24 @@ import java.util.List; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; -class Qos1PublishHandler { +class Qos1PublishHandler extends QosPublishHandler { private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); - private final IAuthorizator m_authorizator; private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; private final ConnectionDescriptorStore connectionDescriptors; - private final String brokerPort; private final MessagesPublisher publisher; public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConnectionDescriptorStore connectionDescriptors, String brokerPort, MessagesPublisher messagesPublisher) { - this.m_authorizator = authorizator; + super(authorizator); this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; this.connectionDescriptors = connectionDescriptors; - this.brokerPort = brokerPort; this.publisher = messagesPublisher; } @@ -53,11 +50,15 @@ class Qos1PublishHandler { String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - LOG.debug("publish2Subscribers_qos1 republishing to existing subscribers that matches the topic {}", topic); - if (LOG.isTraceEnabled()) { - LOG.trace("content <{}>", DebugUtils.payload2Str(toStoreMsg.getMessage())); - LOG.trace("subscription tree {}", subscriptions.dumpTree()); - } + if (LOG.isDebugEnabled()) { + LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", + clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()), + subscriptions.dumpTree()); + } else { + LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID, + topic, msg.getMessageID()); + } + List topicMatchingSubscriptions = subscriptions.matches(topic); this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); @@ -66,7 +67,6 @@ class Qos1PublishHandler { if (msg.isLocal()) { sendPubAck(clientID, messageID); } - LOG.info("server {} replying with PubAck to MSG ID {}", brokerPort, messageID); if (msg.isRetainFlag()) { if (!msg.getPayload().hasRemaining()) { @@ -82,16 +82,6 @@ class Qos1PublishHandler { m_interceptor.notifyTopicPublished(msg, clientID, username); } - boolean checkWriteOnTopic(String topic, Channel channel) { - String clientID = NettyUtils.clientID(channel); - String username = NettyUtils.userName(channel); - if (!m_authorizator.canWrite(topic, username, clientID)) { - LOG.debug("topic {} doesn't have write credentials", topic); - return true; - } - return false; - } - private void sendPubAck(String clientId, int messageID) { LOG.trace("sendPubAck invoked"); PubAckMessage pubAckMessage = new PubAckMessage(); 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 5bc9651a..fe0c59ed 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -14,39 +14,36 @@ import io.netty.channel.Channel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.List; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; -class Qos2PublishHandler { +import java.util.List; + +class Qos2PublishHandler extends QosPublishHandler { private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); - private final IAuthorizator m_authorizator; private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; private final ConnectionDescriptorStore connectionDescriptors; private final ISessionsStore m_sessionsStore; - private final String brokerPort; private final MessagesPublisher publisher; public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, String brokerPort, MessagesPublisher messagesPublisher) { - this.m_authorizator = authorizator; + super(authorizator); this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; - this.brokerPort = brokerPort; this.publisher = messagesPublisher; } void receivedPublishQos2(Channel channel, PublishMessage msg) { - final AbstractMessage.QOSType qos = AbstractMessage.QOSType.EXACTLY_ONCE; final String topic = msg.getTopicName(); //check if the topic can be wrote if (checkWriteOnTopic(topic, channel)) { @@ -57,7 +54,16 @@ class Qos2PublishHandler { IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", this.brokerPort, clientID, topic, qos); + + if (LOG.isDebugEnabled()) { + LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", + clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(toStoreMsg.getMessage()), + subscriptions.dumpTree()); + } else { + LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID, + topic, msg.getMessageID()); + } + //QoS2 MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); if (msg.isLocal()) { @@ -84,16 +90,20 @@ class Qos2PublishHandler { void processPubRel(Channel channel, PubRelMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); - LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); + LOG.info("Processing PUBREL message. MqttClientId = {}, messageId = {}.", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); final String topic = evt.getTopic(); List topicMatchingSubscriptions = subscriptions.matches(topic); - LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic); - if (LOG.isTraceEnabled()) { - LOG.trace("content <{}>", DebugUtils.payload2Str(evt.getMessage())); - LOG.trace("subscription tree {}", subscriptions.dumpTree()); - } + + if (LOG.isDebugEnabled()) { + LOG.debug("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}, payload = {}, subscriptionTree = {}.", + clientID, topic, msg.getMessageID(), DebugUtils.payload2Str(evt.getMessage()), + subscriptions.dumpTree()); + } else { + LOG.info("Sending publish message to subscribers. MqttClientId = {}, topic = {}, messageId = {}.", clientID, + topic, msg.getMessageID()); + } this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions); if (evt.isRetained()) { @@ -107,25 +117,20 @@ class Qos2PublishHandler { sendPubComp(clientID, messageID); } - private boolean checkWriteOnTopic(String topic, Channel channel) { - String clientID = NettyUtils.clientID(channel); - String username = NettyUtils.userName(channel); - if (!m_authorizator.canWrite(topic, username, clientID)) { - LOG.debug("topic {} doesn't have write credentials", topic); - return true; - } - return false; - } private void sendPubRec(String clientID, int messageID) { - LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); + if (LOG.isDebugEnabled()) { + 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) { - LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID); + if (LOG.isDebugEnabled()) { + 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/QosPublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java new file mode 100644 index 00000000..6f5988f4 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java @@ -0,0 +1,30 @@ +package io.moquette.spi.impl; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.moquette.server.netty.NettyUtils; +import io.moquette.spi.security.IAuthorizator; +import io.netty.channel.Channel; + +abstract class QosPublishHandler { + + private static final Logger LOG = LoggerFactory.getLogger(QosPublishHandler.class); + + private final IAuthorizator m_authorizator; + + protected QosPublishHandler(IAuthorizator m_authorizator) { + this.m_authorizator = m_authorizator; + } + + public boolean checkWriteOnTopic(String topic, Channel channel) { + String clientID = NettyUtils.clientID(channel); + String username = NettyUtils.userName(channel); + if (!m_authorizator.canWrite(topic, username, clientID)) { + LOG.error("The MQTT client is not authorized to publish on topic. MqttClientId = {}, topic = {}.", clientID, + topic); + return true; + } + return false; + } +}