From c659a5415891c13240d391a17e2720fea242e1e0 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 20 Mar 2017 21:58:18 +0100 Subject: [PATCH] Minor, just code layout --- .../moquette/spi/impl/ProtocolProcessor.java | 307 ++++++------------ 1 file changed, 94 insertions(+), 213 deletions(-) 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 4b2ce259..a9a8b246 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -19,7 +19,6 @@ package io.moquette.spi.impl; import io.moquette.interception.InterceptHandler; import io.moquette.interception.messages.InterceptAcknowledgedMessage; import io.moquette.server.ConnectionDescriptor; -import io.moquette.server.ConnectionDescriptor.ConnectionState; import io.moquette.server.ConnectionDescriptorStore; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.*; @@ -50,6 +49,7 @@ import static io.moquette.spi.impl.Utils.readBytesAndRewind; import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.*; +import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; /** * Class responsible to handle the logic of MQTT protocol it's the director of the protocol @@ -160,47 +160,22 @@ public class ProtocolProcessor { public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) { - init( - subscriptions, - storageService, - sessionsStore, - authenticator, - allowAnonymous, - false, - authorizator, - interceptor, - null); + init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator, + interceptor, null); } public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor) { - init( - subscriptions, - storageService, - sessionsStore, - authenticator, - allowAnonymous, - allowZeroByteClientId, - authorizator, - interceptor, - null); + init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, + authorizator, interceptor, null); } public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { - init( - new ConnectionDescriptorStore(sessionsStore), - subscriptions, - storageService, - sessionsStore, - authenticator, - allowAnonymous, - allowZeroByteClientId, - authorizator, - interceptor, - serverPort); + init(new ConnectionDescriptorStore(sessionsStore), subscriptions, storageService, sessionsStore, authenticator, + allowAnonymous, allowZeroByteClientId, authorizator, interceptor, serverPort); } /** @@ -234,7 +209,7 @@ public class ProtocolProcessor { this.allowZeroByteClientId = allowZeroByteClientId; m_authorizator = authorizator; if (LOG.isDebugEnabled()) { - LOG.debug("Initial subscriptions tree = {}.", subscriptions.dumpTree()); + LOG.debug("Initial subscriptions tree={}.", subscriptions.dumpTree()); } m_authenticator = authenticator; m_messagesStore = storageService; @@ -243,36 +218,16 @@ public class ProtocolProcessor { LOG.info("Initializing messages publisher..."); final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors); - this.messagesPublisher = new MessagesPublisher( - connectionDescriptors, - sessionsStore, - m_messagesStore, + 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, - m_interceptor, - this.connectionDescriptors, - m_server_port, - this.messagesPublisher); - this.qos2PublishHandler = new Qos2PublishHandler( - m_authorizator, - subscriptions, - m_messagesStore, - m_interceptor, - this.connectionDescriptors, - m_sessionsStore, - m_server_port, + this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor, this.messagesPublisher); + this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor, + this.connectionDescriptors, m_server_port, this.messagesPublisher); + 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); @@ -281,13 +236,13 @@ public class ProtocolProcessor { public void processConnect(Channel channel, MqttConnectMessage msg) { MqttConnectPayload payload = msg.payload(); String clientId = payload.clientIdentifier(); - LOG.info("Processing CONNECT message. CId = {}, username = {}", clientId, payload.userName()); + LOG.info("Processing CONNECT message. CId={}, username={}", clientId, payload.userName()); if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel() && msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()) { MqttConnAckMessage badProto = connAck(CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION); - LOG.error("The MQTT protocol version is not valid. CId = {}", clientId); + LOG.error("MQTT protocol version is not valid. CId={}", clientId); channel.writeAndFlush(badProto); channel.close(); return; @@ -299,16 +254,14 @@ public class ProtocolProcessor { channel.writeAndFlush(badId); channel.close(); - LOG.error("The MQTT client ID cannot be empty. Username = {}", payload.userName()); + LOG.error("The MQTT client ID cannot be empty. Username={}", payload.userName()); return; } // Generating client id. clientId = UUID.randomUUID().toString().replace("-", ""); - LOG.info( - "The client has connected with a server generated identifier. CId = {}, username = {}", - clientId, - payload.userName()); + LOG.info("Client has connected with a server generated identifier. CId={}, username={}", clientId, + payload.userName()); } if (!login(channel, msg, clientId)) { @@ -316,21 +269,17 @@ public class ProtocolProcessor { return; } - ConnectionDescriptor descriptor = new ConnectionDescriptor( - clientId, - channel, - msg.variableHeader().isCleanSession()); + ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, + msg.variableHeader().isCleanSession()); ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor); if (existing != null) { - LOG.info("The client ID is being used in an existing connection. It will be closed. CId = {}", clientId); + LOG.info("The client ID is being used in an existing connection. It will be closed. CId={}", clientId); existing.abort(); return; } initializeKeepAliveTimeout(channel, msg, clientId); - storeWillMessage(msg, clientId); - if (!sendAck(descriptor, msg, clientId)) { channel.close(); return; @@ -348,13 +297,12 @@ public class ProtocolProcessor { channel.close(); return; } - final boolean success = descriptor - .assignState(ConnectionState.MESSAGES_REPUBLISHED, ConnectionState.ESTABLISHED); + final boolean success = descriptor.assignState(MESSAGES_REPUBLISHED, ESTABLISHED); if (!success) { channel.close(); } - LOG.info("The CONNECT message has been processed. CId = {}, username = {}.", clientId, payload.userName()); + LOG.info("The CONNECT message has been processed. CId={}, username={}.", clientId, payload.userName()); } private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) { @@ -366,12 +314,8 @@ public class ProtocolProcessor { } private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode, boolean sessionPresent) { - MqttFixedHeader mqttFixedHeader = new MqttFixedHeader( - MqttMessageType.CONNACK, - false, - MqttQoS.AT_MOST_ONCE, - false, - 0); + MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, + false, 0); MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(returnCode, sessionPresent); return new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader); } @@ -383,26 +327,19 @@ public class ProtocolProcessor { if (msg.variableHeader().hasPassword()) { pwd = msg.payload().password().getBytes(); } else if (!this.allowAnonymous) { - LOG.error( - "The client didn't supply any password and MQTT anonymous mode is disabled. CId = {}.", - clientId); + LOG.error("Client didn't supply any password and MQTT anonymous mode is disabled CId={}", clientId); failedCredentials(channel); return false; } if (!m_authenticator.checkValid(clientId, msg.payload().userName(), pwd)) { - LOG.error( - "The authenticator has rejected the MQTT credentials. CId = {}, username = {}, password = {}.", - clientId, - msg.payload().userName(), - pwd); + LOG.error("Authenticator has rejected the MQTT credentials CId={}, username={}, password={}", + clientId, msg.payload().userName(), pwd); failedCredentials(channel); return false; } NettyUtils.userName(channel, msg.payload().userName()); } else if (!this.allowAnonymous) { - LOG.error( - "The client didn't supply any credentials and MQTT anonymous mode is disabled. CId = {}.", - clientId); + LOG.error("Client didn't supply any credentials and MQTT anonymous mode is disabled. CId={}", clientId); failedCredentials(channel); return false; } @@ -410,8 +347,8 @@ public class ProtocolProcessor { } private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, final String clientId) { - LOG.info("Sending connect ACK. CId = {}.", clientId); - final boolean success = descriptor.assignState(ConnectionState.DISCONNECTED, ConnectionState.SENDACK); + LOG.info("Sending connect ACK. CId={}.", clientId); + final boolean success = descriptor.assignState(DISCONNECTED, SENDACK); if (!success) { return false; } @@ -426,17 +363,17 @@ public class ProtocolProcessor { } if (isSessionAlreadyStored) { - LOG.info("Cleaning session. CId = {}.", clientId); + LOG.info("Cleaning session. CId={}.", clientId); clientSession.cleanSession(msg.variableHeader().isCleanSession()); } descriptor.writeAndFlush(okResp); - LOG.info("The connect ACK has been sent. CId = {}.", clientId); + LOG.info("The connect ACK has been sent. CId={}.", clientId); return true; } private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, final String clientId) { int keepAlive = msg.variableHeader().keepAliveTimeSeconds(); - LOG.info("Configuring connection. CId = {}.", clientId); + LOG.info("Configuring connection. CId={}.", clientId); NettyUtils.keepAlive(channel, keepAlive); // session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.variableHeader().isCleanSession()); NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession()); @@ -446,41 +383,29 @@ public class ProtocolProcessor { int idleTime = Math.round(keepAlive * 1.5f); setIdleTime(channel.pipeline(), idleTime); - LOG.info( - "The connection has been configured. CId = {}, keepAlive = {}, cleanSession = {}, idleTime = {}.", - clientId, - keepAlive, - msg.variableHeader().isCleanSession(), - idleTime); + LOG.debug("The connection has been configured CId={}, keepAlive={}, cleanSession={}, idleTime={}", + clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime); } private void storeWillMessage(MqttConnectMessage msg, final String clientId) { // Handle will flag if (msg.variableHeader().isWillFlag()) { MqttQoS willQos = MqttQoS.valueOf(msg.variableHeader().willQos()); - LOG.info( - "Configuring MQTT last will and testament. " - + "CId = {}, willQos = {}, willTopic = {}, willRetain = {}.", - clientId, - willQos, - msg.payload().willTopic(), - msg.variableHeader().isWillRetain()); + LOG.info("Configuring MQTT last will and testament CId={}, willQos={}, willTopic={}, willRetain={}", + clientId, willQos, msg.payload().willTopic(), msg.variableHeader().isWillRetain()); byte[] willPayload = msg.payload().willMessage().getBytes(); ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); // save the will testament in the clientID store - WillMessage will = new WillMessage( - msg.payload().willTopic(), - bb, - msg.variableHeader().isWillRetain(), + WillMessage will = new WillMessage(msg.payload().willTopic(), bb, msg.variableHeader().isWillRetain(), willQos); m_willStore.put(clientId, will); - LOG.info("MQTT last will and testament has been configured. CId = {}.", clientId); + LOG.info("MQTT last will and testament has been configured. CId={}.", clientId); } } private ClientSession createOrLoadClientSession(ConnectionDescriptor descriptor, MqttConnectMessage msg, String clientId) { - final boolean success = descriptor.assignState(ConnectionState.SENDACK, ConnectionState.SESSION_CREATED); + final boolean success = descriptor.assignState(SENDACK, SESSION_CREATED); if (!success) { return null; } @@ -491,15 +416,14 @@ public class ProtocolProcessor { clientSession = m_sessionsStore.createNewSession(clientId, msg.variableHeader().isCleanSession()); } if (msg.variableHeader().isCleanSession()) { - LOG.info("Cleaning session. CId = {}.", clientId); + LOG.info("Cleaning session. CId={}.", clientId); clientSession.cleanSession(); } return clientSession; } private boolean republish(ConnectionDescriptor descriptor, MqttConnectMessage msg, ClientSession clientSession) { - final boolean success = descriptor - .assignState(ConnectionState.SESSION_CREATED, ConnectionState.MESSAGES_REPUBLISHED); + final boolean success = descriptor.assignState(SESSION_CREATED, MESSAGES_REPUBLISHED); if (!success) { return false; } @@ -529,10 +453,10 @@ public class ProtocolProcessor { * Republish QoS1 and QoS2 messages stored into the session for the clientID. */ private void republishStoredInSession(ClientSession clientSession) { - LOG.info("Republishing stored publish events. CId = {}.", clientSession.clientID); + LOG.info("Republishing stored publish events. CId={}.", clientSession.clientID); BlockingQueue publishedEvents = clientSession.queue(); if (publishedEvents.isEmpty()) { - LOG.info("There are no stored publish events. ClientId = {}.", clientSession.clientID); + LOG.info("There are no stored publish events. ClientId={}.", clientSession.clientID); return; } @@ -558,19 +482,15 @@ public class ProtocolProcessor { ByteBuf payload = msg.payload(); byte[] payloadContent = readBytesAndRewind(payload); - IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage( - payloadContent, - msg.fixedHeader().qosLevel(), - msg.variableHeader().topicName()); + IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(payloadContent, + msg.fixedHeader().qosLevel(), msg.variableHeader().topicName()); stored.setRetained(msg.fixedHeader().isRetain()); stored.setMessageID(msg.variableHeader().messageId()); return stored; } private static IMessagesStore.StoredMessage asStoredMessage(WillMessage will) { - IMessagesStore.StoredMessage pub = new IMessagesStore.StoredMessage( - will.getPayload().array(), - will.getQos(), + IMessagesStore.StoredMessage pub = new IMessagesStore.StoredMessage(will.getPayload().array(), will.getQos(), will.getTopic()); pub.setRetained(will.isRetained()); return pub; @@ -579,12 +499,8 @@ public class ProtocolProcessor { public void processPublish(Channel channel, MqttPublishMessage msg) { final MqttQoS qos = msg.fixedHeader().qosLevel(); final String clientId = NettyUtils.clientID(channel); - LOG.info( - "Processing PUBLISH message. CId = {}, topic = {}, messageId = {}, qos = {}.", - clientId, - msg.variableHeader().topicName(), - msg.variableHeader().messageId(), - qos); + LOG.info("Processing PUBLISH message. CId={}, topic={}, messageId={}, qos={}", clientId, + msg.variableHeader().topicName(), msg.variableHeader().messageId(), qos); switch (qos) { case AT_MOST_ONCE: this.qos0PublishHandler.receivedPublishQos0(channel, msg); @@ -615,7 +531,7 @@ public class ProtocolProcessor { public void internalPublish(MqttPublishMessage msg, final String clientId) { final MqttQoS qos = msg.fixedHeader().qosLevel(); final Topic topic = new Topic(msg.variableHeader().topicName()); - LOG.info("Sending PUBLISH message. Topic = {}, qos = {}.", topic, qos); + LOG.info("Sending PUBLISH message. Topic={}, qos={}.", topic, qos); MessageGUID guid = null; IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); @@ -663,7 +579,7 @@ public class ProtocolProcessor { Topic topic = new Topic(tobeStored.getTopic()); List topicMatchingSubscriptions = subscriptions.matches(topic); - LOG.info("Publishing will message. CId = {}, messageId = {}, topic = {}.", clientID, messageId, topic); + LOG.info("Publishing will message. CId={}, messageId={}, topic={}.", clientID, messageId, topic); this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions); } @@ -694,7 +610,7 @@ public class ProtocolProcessor { int messageID = messageId(msg); targetSession.moveInFlightToSecondPhaseAckWaiting(messageID); // once received a PUBREC reply with a PUBREL(messageID) - LOG.debug("Processing PUBREC message. CId = {}, messageId = {}.", clientID, messageID); + LOG.debug("Processing PUBREC message. CId={}, messageId={}.", clientID, messageID); MqttFixedHeader pubRelHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0); MqttMessage pubRelMessage = new MqttMessage(pubRelHeader, from(messageID)); @@ -704,7 +620,7 @@ public class ProtocolProcessor { public void processPubComp(Channel channel, MqttMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); - LOG.debug("Processing PUBCOMP message. CId = {}, messageId = {}.", clientID, messageID); + LOG.debug("Processing PUBCOMP message. CId={}, 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); @@ -715,7 +631,7 @@ public class ProtocolProcessor { public void processDisconnect(Channel channel) throws InterruptedException { final String clientID = NettyUtils.clientID(channel); - LOG.info("Processing DISCONNECT message. CId = {}.", clientID); + LOG.info("Processing DISCONNECT message. CId={}.", clientID); channel.flush(); final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID); if (existingDescriptor == null) { @@ -726,82 +642,79 @@ public class ProtocolProcessor { if (existingDescriptor.doesNotUseChannel(channel)) { // another client saved it's descriptor, exit - LOG.warn("Another client is using the connection descriptor. Closing connection. CId = {}.", clientID); + LOG.warn("Another client is using the connection descriptor. Closing connection. CId={}.", clientID); existingDescriptor.abort(); return; } if (!removeSubscriptions(existingDescriptor, clientID)) { - LOG.warn("Unable to remove subscriptions. Closing connection. CId = {}.", clientID); + LOG.warn("Unable to remove subscriptions. Closing connection. CId={}.", clientID); existingDescriptor.abort(); return; } if (!dropStoredMessages(existingDescriptor, clientID)) { - LOG.warn("Unable to drop stored messages. Closing connection. CId = {}.", clientID); + LOG.warn("Unable to drop stored messages. Closing connection. CId={}.", clientID); existingDescriptor.abort(); return; } if (!cleanWillMessageAndNotifyInterceptor(existingDescriptor, clientID)) { - LOG.warn("Unable to drop will message. Closing connection. CId = {}.", clientID); + LOG.warn("Unable to drop will message. Closing connection. CId={}.", clientID); existingDescriptor.abort(); return; } if (!existingDescriptor.close()) { - LOG.info("The connection has been closed. CId = {}.", clientID); + LOG.info("The connection has been closed. CId={}.", clientID); return; } boolean stillPresent = this.connectionDescriptors.removeConnection(existingDescriptor); if (!stillPresent) { // another descriptor was inserted - LOG.warn("Another descriptor has been inserted. CId = {}.", clientID); + LOG.warn("Another descriptor has been inserted. CId={}.", clientID); return; } - LOG.info("The DISCONNECT message has been processed. CId = {}.", clientID); + LOG.info("The DISCONNECT message has been processed. CId={}.", clientID); } private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) { - final boolean success = descriptor - .assignState(ConnectionState.ESTABLISHED, ConnectionState.SUBSCRIPTIONS_REMOVED); + final boolean success = descriptor.assignState(ESTABLISHED, SUBSCRIPTIONS_REMOVED); if (!success) { return false; } if (descriptor.cleanSession) { - LOG.info("Removing saved subscriptions. CId = {}.", descriptor.clientID); + LOG.info("Removing saved subscriptions. CId={}.", descriptor.clientID); m_sessionsStore.wipeSubscriptions(clientID); - LOG.info("The saved subscriptions have been removed. CId = {}.", descriptor.clientID); + LOG.info("The saved subscriptions have been removed. CId={}.", descriptor.clientID); } return true; } private boolean dropStoredMessages(ConnectionDescriptor descriptor, String clientID) { - final boolean success = descriptor - .assignState(ConnectionState.SUBSCRIPTIONS_REMOVED, ConnectionState.MESSAGES_DROPPED); + final boolean success = descriptor.assignState(SUBSCRIPTIONS_REMOVED, MESSAGES_DROPPED); if (!success) { return false; } if (descriptor.cleanSession) { - LOG.debug("Removing messages of session. CId = {}.", descriptor.clientID); + LOG.debug("Removing messages of session. CId={}.", descriptor.clientID); this.m_sessionsStore.dropQueue(clientID); - LOG.debug("The messages of the session have been removed. CId = {}.", descriptor.clientID); + LOG.debug("The messages of the session have been removed. CId={}.", descriptor.clientID); } return true; } private boolean cleanWillMessageAndNotifyInterceptor(ConnectionDescriptor descriptor, String clientID) { - final boolean success = descriptor - .assignState(ConnectionState.MESSAGES_DROPPED, ConnectionState.INTERCEPTORS_NOTIFIED); + final boolean success = descriptor.assignState(MESSAGES_DROPPED, INTERCEPTORS_NOTIFIED); if (!success) { return false; } - LOG.info("Removing will message. ClientId = {}.", descriptor.clientID); + LOG.info("Removing will message. ClientId={}.", descriptor.clientID); // cleanup the will store m_willStore.remove(clientID); String username = descriptor.getUsername(); @@ -810,7 +723,7 @@ public class ProtocolProcessor { } public void processConnectionLost(String clientID, Channel channel) { - LOG.info("Processing connection lost event. CId = {}.", clientID); + LOG.info("Processing connection lost event. CId={}.", clientID); ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true); connectionDescriptors.removeConnection(oldConnDescr); // publish the Will message (if any) for the clientID @@ -837,7 +750,7 @@ public class ProtocolProcessor { List topics = msg.payload().topics(); String clientID = NettyUtils.clientID(channel); - LOG.info("Processing UNSUBSCRIBE message. CId = {}, topics = {}.", clientID, topics); + LOG.info("Processing UNSUBSCRIBE message. CId={}, topics={}.", clientID, topics); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); for (String t : topics) { @@ -846,15 +759,12 @@ public class ProtocolProcessor { if (!validTopic) { // close the connection, not valid topicFilter is a protocol violation channel.close(); - LOG.error( - "The topic filter is not valid. CId = {}, topics = {}, badTopicFilter = {}.", - clientID, - topics, - topic); + LOG.error( "Topic filter is not valid. CId={}, topics={}, badTopicFilter={}", clientID, topics, + topic); return; } - LOG.debug("Removing subscription. CId = {}, topic = {}.", clientID, topic); + LOG.debug("Removing subscription. CId={}, topic={}.", clientID, topic); subscriptions.removeSubscription(topic, clientID); clientSession.unsubscribeFrom(topic); String username = NettyUtils.userName(channel); @@ -866,35 +776,29 @@ public class ProtocolProcessor { MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, AT_LEAST_ONCE, false, 0); MqttUnsubAckMessage ackMessage = new MqttUnsubAckMessage(fixedHeader, from(messageID)); - LOG.info("Sending UNSUBACK message. CId = {}, topics = {}, messageId = {}.", clientID, topics, messageID); + LOG.info("Sending UNSUBACK message. CId={}, topics={}, messageId={}.", clientID, topics, messageID); channel.writeAndFlush(ackMessage); } public void processSubscribe(Channel channel, MqttSubscribeMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); - LOG.info("Processing SUBSCRIBE message. CId = {}, messageId = {}.", clientID, messageID); + LOG.info("Processing SUBSCRIBE message. CId={}, messageId={}.", clientID, messageID); RunningSubscription executionKey = new RunningSubscription(clientID, messageID); SubscriptionState currentStatus = this.subscriptionInCourse .putIfAbsent(executionKey, SubscriptionState.VERIFIED); if (currentStatus != null) { - LOG.warn( - "The client sent another SUBSCRIBE message while this one was being processed. " - + "CId = {}, messageId = {}.", - clientID, - messageID); + LOG.warn("Client sent another SUBSCRIBE message while this one was being processed " + + "CId={}, messageId={}", clientID, messageID); return; } String username = NettyUtils.userName(channel); List ackTopics = doVerify(clientID, username, msg); MqttSubAckMessage ackMessage = doAckMessageFromValidateFilters(ackTopics, messageID); if (!this.subscriptionInCourse.replace(executionKey, SubscriptionState.VERIFIED, SubscriptionState.STORED)) { - LOG.warn( - "The client sent another SUBSCRIBE message while the topic filters were being verified. " - + "CId = {}, messageId = {}.", - clientID, - messageID); + LOG.warn("Client sent another SUBSCRIBE message while the topic filters were being verified " + + "CId={}, messageId={}", clientID, messageID); return; } @@ -903,7 +807,6 @@ public class ProtocolProcessor { List newSubscriptions = doStoreSubscription(ackTopics, clientID); // save session, persist subscriptions from session - for (Subscription subscription : newSubscriptions) { subscriptions.add(subscription.asClientTopicCouple()); } @@ -958,32 +861,19 @@ public class ProtocolProcessor { Topic topic = new Topic(req.topicName()); if (!m_authorizator.canRead(topic, username, clientSession.clientID)) { // send SUBACK with 0x80, the user hasn't credentials to read the topic - LOG.error( - "The client does not have read permissions on the topic. " - + "CId = {}, username = {}, messageId = {}, topic = {}.", - clientID, - username, - messageId, - topic); + LOG.error("Client does not have read permissions on the topic " + + "CId={}, username={}, messageId={}, topic={}", + clientID, username, messageId, topic); ackTopics.add(new MqttTopicSubscription(topic.toString(), FAILURE)); } else { MqttQoS qos; if (topic.isValid()) { - LOG.info( - "The client will be subscribed to the topic. " - + "CId = {}, username = {}, messageId = {}, topic = {}.", - clientID, - username, - messageId, - topic); + LOG.info("Client will be subscribed to the topic " + + "CId={}, username={}, messageId={}, topic={}", clientID, username, messageId, topic); qos = req.qualityOfService(); } else { - LOG.error( - "The topic filter is not valid. CId = {}, username = {}, messageId = {}, topic = {}.", - clientID, - username, - messageId, - topic); + LOG.error("Topic filter is not valid CId={}, username={}, messageId={}, topic={}", + clientID, username, messageId, topic); qos = FAILURE; } ackTopics.add(new MqttTopicSubscription(topic.toString(), qos)); @@ -1007,9 +897,7 @@ public class ProtocolProcessor { } private void publishRetainedMessagesInSession(final Subscription newSubscription, String username) { - LOG.info( - "Retrieving retained messages. CId = {}, topics = {}.", - newSubscription.getClientId(), + LOG.info("Retrieving retained messages CId={}, topics={}", newSubscription.getClientId(), newSubscription.getTopicFilter()); // scans retained messages to be published to the new subscription @@ -1023,11 +911,8 @@ public class ProtocolProcessor { }); if (!messages.isEmpty()) { - LOG.info( - "Publishing retained messages. CId = {}, topics = {}, messagesNo = {}.", - newSubscription.getClientId(), - newSubscription.getTopicFilter(), - messages.size()); + LOG.info("Publishing retained messages CId={}, topics={}, messagesNo={}", + newSubscription.getClientId(), newSubscription.getTopicFilter(), messages.size()); } ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId()); this.internalRepublisher.publishRetained(targetSession, messages); @@ -1047,12 +932,8 @@ public class ProtocolProcessor { } else { // recreate a publish from stored publish in queue boolean retained = m_messagesStore.getMessageByGuid(msg.getGuid()) != null; - MqttPublishMessage pubMsg = createPublishForQos( - msg.getTopic(), - msg.getQos(), - msg.getMessage(), - retained, - 0); + MqttPublishMessage pubMsg = createPublishForQos( msg.getTopic(), msg.getQos(), msg.getMessage(), + retained, 0); channel.write(pubMsg); } }