From db31f77f9b6bea2d2e36868213b42e260df90984 Mon Sep 17 00:00:00 2001 From: Andreas Mager Date: Wed, 1 Feb 2017 15:53:32 +0100 Subject: [PATCH] broker/ProtocolProcessor: code formatter used --- .../moquette/spi/impl/ProtocolProcessor.java | 512 +++++++++++------- 1 file changed, 324 insertions(+), 188 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 38a17028..d8f33250 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -13,6 +13,7 @@ * * You may elect to redistribute this code under either of these licenses. */ + package io.moquette.spi.impl; import io.moquette.interception.InterceptHandler; @@ -34,7 +35,6 @@ import io.netty.handler.codec.mqtt.*; import io.netty.handler.timeout.IdleStateHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; @@ -43,7 +43,6 @@ import java.util.UUID; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; - import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos; import static io.moquette.spi.impl.Utils.messageId; import static io.moquette.spi.impl.Utils.readBytesAndRewind; @@ -52,8 +51,8 @@ import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.*; /** - * Class responsible to handle the logic of MQTT protocol it's the director of - * the protocol execution. + * Class responsible to handle the logic of MQTT protocol it's the director of the protocol + * execution. * * Used by the front facing class ProtocolProcessorBootstrapper. * @@ -62,6 +61,7 @@ import static io.netty.handler.codec.mqtt.MqttQoS.*; public class ProtocolProcessor { static final class WillMessage { + private final String topic; private final ByteBuffer payload; private final boolean retained; @@ -89,7 +89,6 @@ public class ProtocolProcessor { public MqttQoS getQos() { return qos; } - } private enum SubscriptionState { @@ -97,6 +96,7 @@ public class ProtocolProcessor { } private class RunningSubscription { + final String clientID; final long packetId; @@ -107,12 +107,15 @@ public class ProtocolProcessor { @Override public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; + if (this == o) + return true; + if (o == null || getClass() != o.getClass()) + return false; RunningSubscription that = (RunningSubscription) o; - return packetId == that.packetId && (clientID != null ? clientID.equals(that.clientID) : that.clientID == null); + return packetId == that.packetId + && (clientID != null ? clientID.equals(that.clientID) : that.clientID == null); } @@ -145,52 +148,80 @@ public class ProtocolProcessor { private MessagesPublisher messagesPublisher; private InternalRepublisher internalRepublisher; - //maps clientID to Will testament, if specified on CONNECT + // maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); - 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); + ProtocolProcessor() { } - 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); + 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); + } + + 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); + } + + 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); } - - 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); - } /** - * @param subscriptions the subscription store where are stored all the existing - * clients subscriptions. - * @param storageService the persistent store to use for save/load of messages - * for QoS1 and QoS2 handling. - * @param sessionsStore the clients sessions store, used to persist subscriptions. - * @param authenticator the authenticator used in connect messages. - * @param allowAnonymous true connection to clients without credentials. - * @param allowZeroByteClientId true to allow clients connect without a clientid - * @param authorizator used to apply ACL policies to publishes and subscriptions. - * @param interceptor to notify events to an intercept handler + * @param subscriptions + * the subscription store where are stored all the existing clients subscriptions. + * @param storageService + * the persistent store to use for save/load of messages for QoS1 and QoS2 handling. + * @param sessionsStore + * the clients sessions store, used to persist subscriptions. + * @param authenticator + * the authenticator used in connect messages. + * @param allowAnonymous + * true connection to clients without credentials. + * @param allowZeroByteClientId + * true to allow clients connect without a clientid + * @param authorizator + * used to apply ACL policies to publishes and subscriptions. + * @param interceptor + * to notify events to an intercept handler */ - void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsStore subscriptions, IMessagesStore storageService, - ISessionsStore sessionsStore, - IAuthenticator authenticator, - boolean allowAnonymous, - boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { + void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsStore subscriptions, + IMessagesStore storageService, ISessionsStore sessionsStore, 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<>(); @@ -199,9 +230,9 @@ public class ProtocolProcessor { this.allowAnonymous = allowAnonymous; this.allowZeroByteClientId = allowZeroByteClientId; m_authorizator = authorizator; - if (LOG.isDebugEnabled()) { - LOG.debug("Initial subscriptions tree = {}.", subscriptions.dumpTree()); - } + if (LOG.isDebugEnabled()) { + LOG.debug("Initial subscriptions tree = {}.", subscriptions.dumpTree()); + } m_authenticator = authenticator; m_messagesStore = storageService; m_sessionsStore = sessionsStore; @@ -209,15 +240,36 @@ public class ProtocolProcessor { LOG.info("Initializing messages publisher..."); final PersistentQueueMessageSender messageSender = new PersistentQueueMessageSender(this.connectionDescriptors); - this.messagesPublisher = new MessagesPublisher(connectionDescriptors, sessionsStore, m_messagesStore, messageSender); + 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.messagesPublisher); + 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); @@ -228,8 +280,8 @@ public class ProtocolProcessor { String clientId = payload.clientIdentifier(); 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()) { + 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); @@ -250,8 +302,10 @@ public class ProtocolProcessor { // 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( + "The client has connected with a server generated identifier. CId = {}, username = {}", + clientId, + payload.userName()); } if (!login(channel, msg, clientId)) { @@ -259,11 +313,13 @@ 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; } @@ -289,7 +345,8 @@ public class ProtocolProcessor { channel.close(); return; } - final boolean success = descriptor.assignState(ConnectionState.MESSAGES_REPUBLISHED, ConnectionState.ESTABLISHED); + final boolean success = descriptor + .assignState(ConnectionState.MESSAGES_REPUBLISHED, ConnectionState.ESTABLISHED); if (!success) { channel.close(); } @@ -306,32 +363,42 @@ 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); } private boolean login(Channel channel, MqttConnectMessage msg, final String clientId) { - //handle user authentication + // handle user authentication if (msg.variableHeader().hasUserName()) { byte[] pwd = null; 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 = {}.", + LOG.error( + "The 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( + "The 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 = {}.", + LOG.error( + "The client didn't supply any credentials and MQTT anonymous mode is disabled. CId = {}.", clientId); failedCredentials(channel); return false; @@ -340,7 +407,7 @@ public class ProtocolProcessor { } private boolean sendAck(ConnectionDescriptor descriptor, MqttConnectMessage msg, final String clientId) { - LOG.info("Sending connect ACK. CId = {}.", clientId); + LOG.info("Sending connect ACK. CId = {}.", clientId); final boolean success = descriptor.assignState(ConnectionState.DISCONNECTED, ConnectionState.SENDACK); if (!success) { return false; @@ -356,7 +423,7 @@ 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); @@ -368,35 +435,48 @@ public class ProtocolProcessor { int keepAlive = msg.variableHeader().keepAliveTimeSeconds(); LOG.info("Configuring connection. CId = {}.", clientId); NettyUtils.keepAlive(channel, keepAlive); - //session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.variableHeader().isCleanSession()); + // session.attr(NettyUtils.ATTR_KEY_CLEANSESSION).set(msg.variableHeader().isCleanSession()); NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession()); - //used to track the client in the subscription and publishing phases. - //session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); + // used to track the client in the subscription and publishing phases. + // session.attr(NettyUtils.ATTR_KEY_CLIENTID).set(msg.getClientID()); NettyUtils.clientID(channel, clientId); 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.info( + "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 + // 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(), willQos); + // save the will testament in the clientID store + 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); } } private ClientSession createOrLoadClientSession(ConnectionDescriptor descriptor, MqttConnectMessage msg, - String clientId) { + String clientId) { final boolean success = descriptor.assignState(ConnectionState.SENDACK, ConnectionState.SESSION_CREATED); if (!success) { return null; @@ -408,23 +488,24 @@ 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(ConnectionState.SESSION_CREATED, ConnectionState.MESSAGES_REPUBLISHED); if (!success) { return false; } if (!msg.variableHeader().isCleanSession()) { - //force the republish of stored QoS1 and QoS2 + // force the republish of stored QoS1 and QoS2 republishStoredInSession(clientSession); } - int flushIntervalMs = 500/*(keepAlive * 1000) / 2*/; + int flushIntervalMs = 500/* (keepAlive * 1000) / 2 */; descriptor.setupAutoFlusher(flushIntervalMs); return true; } @@ -443,7 +524,7 @@ 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); BlockingQueue publishedEvents = clientSession.queue(); @@ -470,19 +551,24 @@ public class ProtocolProcessor { } public static IMessagesStore.StoredMessage asStoredMessage(MqttPublishMessage msg) { - //TODO ugly, too much array copy + // TODO ugly, too much array copy 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(), will.getTopic()); + IMessagesStore.StoredMessage pub = new IMessagesStore.StoredMessage( + will.getPayload().array(), + will.getQos(), + will.getTopic()); pub.setRetained(will.isRetained()); return pub; } @@ -490,8 +576,12 @@ 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); @@ -503,21 +593,22 @@ public class ProtocolProcessor { this.qos2PublishHandler.receivedPublishQos2(channel, msg); break; default: - break; + break; } } /** - * Intended usage is only for embedded versions of the broker, where the hosting application want to use the - * broker to send a publish message. - * Inspired by {@link #processPublish} but with some changes to avoid security check, and the handshake phases - * for Qos1 and Qos2. - * It also doesn't notifyTopicPublished because using internally the owner should already know where - * it's publishing. + * Intended usage is only for embedded versions of the broker, where the hosting application + * want to use the broker to send a publish message. Inspired by {@link #processPublish} but + * with some changes to avoid security check, and the handshake phases for Qos1 and Qos2. It + * also doesn't notifyTopicPublished because using internally the owner should already know + * where it's publishing. * - * @param msg the message to publish. - * @param clientId the clientID - * */ + * @param msg + * the message to publish. + * @param clientId + * the clientID + */ public void internalPublish(MqttPublishMessage msg, final String clientId) { final MqttQoS qos = msg.fixedHeader().qosLevel(); final String topic = msg.variableHeader().topicName(); @@ -531,7 +622,7 @@ public class ProtocolProcessor { toStoreMsg.setClientID(clientId); } toStoreMsg.setMessageID(1); - if (qos == EXACTLY_ONCE) { //QoS2 + if (qos == EXACTLY_ONCE) { // QoS2 guid = m_messagesStore.storePublishForFuture(toStoreMsg); } List topicMatchingSubscriptions = subscriptions.matches(topic); @@ -541,12 +632,12 @@ public class ProtocolProcessor { return; } if (qos == AT_MOST_ONCE || msg.payload().readableBytes() == 0) { - //QoS == 0 && retain => clean old retained + // QoS == 0 && retain => clean old retained m_messagesStore.cleanRetained(topic); return; } if (guid == null) { - //before wasn't stored + // before wasn't stored guid = m_messagesStore.storePublishForFuture(toStoreMsg); } m_messagesStore.storeRetained(topic, guid); @@ -556,8 +647,8 @@ public class ProtocolProcessor { * Specialized version to publish will testament message. */ private void forwardPublishWill(WillMessage will, String clientID) { - //it has just to publish the message downstream to the subscribers - //NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1 + // it has just to publish the message downstream to the subscribers + // NB it's a will publish, it needs a PacketIdentifier for this conn, default to 1 Integer messageId = null; if (will.getQos() != AT_MOST_ONCE) { messageId = m_sessionsStore.nextPacketID(clientID); @@ -569,7 +660,7 @@ public class ProtocolProcessor { String 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); } @@ -581,11 +672,14 @@ public class ProtocolProcessor { } /** - * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish - * to all interested subscribers. - * @param channel the channel of the incoming message. - * @param msg the decoded pubrel message. - * */ + * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored + * message and publish to all interested subscribers. + * + * @param channel + * the channel of the incoming message. + * @param msg + * the decoded pubrel message. + */ public void processPubRel(Channel channel, MqttMessage msg) { this.qos2PublishHandler.processPubRel(channel, msg); } @@ -593,10 +687,10 @@ public class ProtocolProcessor { public void processPubRec(Channel channel, MqttMessage msg) { String clientID = NettyUtils.clientID(channel); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - //remove from the inflight and move to the QoS2 second phase queue + // remove from the inflight and move to the QoS2 second phase queue int messageID = messageId(msg); targetSession.moveInFlightToSecondPhaseAckWaiting(messageID); - //once received a PUBREC reply with a PUBREL(messageID) + // once received a PUBREC reply with a PUBREL(messageID) LOG.debug("Processing PUBREC message. CId = {}, messageId = {}.", clientID, messageID); MqttFixedHeader pubRelHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0); @@ -608,7 +702,7 @@ public class ProtocolProcessor { String clientID = NettyUtils.clientID(channel); int messageID = messageId(msg); LOG.debug("Processing PUBCOMP message. CId = {}, messageId = {}.", clientID, messageID); - //once received the PUBCOMP then remove the message from the temp memory + // once received the PUBCOMP then remove the message from the temp memory ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID); String username = NettyUtils.userName(channel); @@ -617,59 +711,59 @@ public class ProtocolProcessor { } public void processDisconnect(Channel channel) throws InterruptedException { - final String clientID = NettyUtils.clientID(channel); - LOG.info("Processing DISCONNECT message. CId = {}.", clientID); - channel.flush(); + final String clientID = NettyUtils.clientID(channel); + LOG.info("Processing DISCONNECT message. CId = {}.", clientID); + channel.flush(); final ConnectionDescriptor existingDescriptor = this.connectionDescriptors.getConnection(clientID); if (existingDescriptor == null) { - //another client with same ID removed the descriptor, we must exit + // another client with same ID removed the descriptor, we must exit channel.close(); return; } if (existingDescriptor.doesNotUseChannel(channel)) { - //another client saved it's descriptor, exit - LOG.warn("Another client is using the connection descriptor. Closing connection. CId = {}.", - clientID); - existingDescriptor.abort(); + // another client saved it's descriptor, exit + 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); - existingDescriptor.abort(); + 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); - existingDescriptor.abort(); + 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); - existingDescriptor.abort(); + 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); - return; + // another descriptor was inserted + 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(ConnectionState.ESTABLISHED, ConnectionState.SUBSCRIPTIONS_REMOVED); if (!success) { return false; } @@ -683,7 +777,8 @@ public class ProtocolProcessor { } private boolean dropStoredMessages(ConnectionDescriptor descriptor, String clientID) { - final boolean success = descriptor.assignState(ConnectionState.SUBSCRIPTIONS_REMOVED, ConnectionState.MESSAGES_DROPPED); + final boolean success = descriptor + .assignState(ConnectionState.SUBSCRIPTIONS_REMOVED, ConnectionState.MESSAGES_DROPPED); if (!success) { return false; } @@ -697,13 +792,14 @@ public class ProtocolProcessor { } private boolean cleanWillMessageAndNotifyInterceptor(ConnectionDescriptor descriptor, String clientID) { - final boolean success = descriptor.assignState(ConnectionState.MESSAGES_DROPPED, ConnectionState.INTERCEPTORS_NOTIFIED); + final boolean success = descriptor + .assignState(ConnectionState.MESSAGES_DROPPED, ConnectionState.INTERCEPTORS_NOTIFIED); if (!success) { return false; } LOG.info("Removing will message. ClientId = {}.", descriptor.clientID); - //cleanup the will store + // cleanup the will store m_willStore.remove(clientID); String username = descriptor.getUsername(); m_interceptor.notifyClientDisconnected(clientID, username); @@ -711,10 +807,10 @@ 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 + // publish the Will message (if any) for the clientID if (m_willStore.containsKey(clientID)) { WillMessage will = m_willStore.get(clientID); forwardPublishWill(will, clientID); @@ -726,25 +822,31 @@ public class ProtocolProcessor { } /** - * Remove the clientID from topic subscription, if not previously subscribed, - * doesn't reply any error. - * @param channel the channel of the incoming message. - * @param msg the decoded unsubscribe message. + * Remove the clientID from topic subscription, if not previously subscribed, doesn't reply any + * error. + * + * @param channel + * the channel of the incoming message. + * @param msg + * the decoded unsubscribe message. */ public void processUnsubscribe(Channel channel, MqttUnsubscribeMessage msg) { 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 topic : topics) { boolean validTopic = SubscriptionsStore.validate(topic); if (!validTopic) { - //close the connection, not valid topicFilter is a protocol violation + // 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( + "The topic filter is not valid. CId = {}, topics = {}, badTopicFilter = {}.", + clientID, + topics, + topic); return; } @@ -755,57 +857,64 @@ public class ProtocolProcessor { m_interceptor.notifyTopicUnsubscribed(topic, clientID, username); } - //ack the client + // ack the client int messageID = msg.variableHeader().messageId(); 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); + 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( + "The 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( + "The client sent another SUBSCRIBE message while the topic filters were being verified. " + + "CId = {}, messageId = {}.", + clientID, + messageID); return; } - LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics); + LOG.info("Creating and storing subscriptions CId={}, messageId={}, topics={}", clientID, messageID, ackTopics); List newSubscriptions = doStoreSubscription(ackTopics, clientID); - //save session, persist subscriptions from session + // save session, persist subscriptions from session for (Subscription subscription : newSubscriptions) { subscriptions.add(subscription.asClientTopicCouple()); } - LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID); + LOG.info("Sending SUBACK response CId={}, messageId={}", clientID, messageID); channel.writeAndFlush(ackMessage); - //fire the persisted messages in session + // fire the persisted messages in session for (Subscription subscription : newSubscriptions) { publishRetainedMessagesInSession(subscription, username); } boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); if (!success) { - LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID); + LOG.warn("Unable to perform the final subscription state update CId={}, messageId={}", clientID, messageID); } } @@ -814,7 +923,7 @@ public class ProtocolProcessor { List newSubscriptions = new ArrayList<>(); for (MqttTopicSubscription req : ackTopics) { - //TODO this is SUPER UGLY + // TODO this is SUPER UGLY if (req.qualityOfService() == FAILURE) { continue; } @@ -826,11 +935,14 @@ public class ProtocolProcessor { } /** - * @param clientID the clientID - * @param username the username - * @param msg the subscribe message to verify + * @param clientID + * the clientID + * @param username + * the username + * @param msg + * the subscribe message to verify * @return the list of verified topics for the given subscribe message. - * */ + */ private List doVerify(String clientID, String username, MqttSubscribeMessage msg) { ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); List ackTopics = new ArrayList<>(); @@ -838,20 +950,33 @@ public class ProtocolProcessor { final int messageId = messageId(msg); for (MqttTopicSubscription req : msg.payload().topicSubscriptions()) { if (!m_authorizator.canRead(req.topicName(), 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, req.topicName()); + // 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, + req.topicName()); ackTopics.add(new MqttTopicSubscription(req.topicName(), FAILURE)); } else { MqttQoS qos; if (SubscriptionsStore.validate(req.topicName())) { - LOG.info("The client will be subscribed to the topic. CId = {}, username = {}, messageId = {}, topic = {}.", - clientID, username, messageId, req.topicName()); + LOG.info( + "The client will be subscribed to the topic. " + + "CId = {}, username = {}, messageId = {}, topic = {}.", + clientID, + username, + messageId, + req.topicName()); qos = req.qualityOfService(); } else { - LOG.error("The topic filter is not valid. CId = {}, username = {}, messageId = {}, topic = {}.", - clientID, username, messageId, req.topicName()); - qos = FAILURE; + LOG.error( + "The topic filter is not valid. CId = {}, username = {}, messageId = {}, topic = {}.", + clientID, + username, + messageId, + req.topicName()); + qos = FAILURE; } ackTopics.add(new MqttTopicSubscription(req.topicName(), qos)); } @@ -861,7 +986,7 @@ public class ProtocolProcessor { /** * Create the SUBACK response from a list of topicFilters - * */ + */ private MqttSubAckMessage doAckMessageFromValidateFilters(List topicFilters, int messageId) { List grantedQoSLevels = new ArrayList<>(); for (MqttTopicSubscription req : topicFilters) { @@ -874,12 +999,15 @@ public class ProtocolProcessor { } private void publishRetainedMessagesInSession(final Subscription newSubscription, String username) { - LOG.info("Retrieving retained messages. CId = {}, topics = {}.", newSubscription.getClientId(), - newSubscription.getTopicFilter()); + LOG.info( + "Retrieving retained messages. CId = {}, 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 + // scans retained messages to be published to the new subscription + // TODO this is ugly, it does a linear scan on potential big dataset Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { + @Override public boolean match(String key) { return SubscriptionsStore.matchTopics(key, newSubscription.getTopicFilter()); @@ -887,13 +1015,16 @@ 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); - //notify the Observables + // notify the Observables m_interceptor.notifyTopicSubscribed(newSubscription, username); } @@ -901,14 +1032,19 @@ public class ProtocolProcessor { String clientID = NettyUtils.clientID(channel); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); boolean emptyQueue = false; - while (channel.isWritable() && !emptyQueue) { + while (channel.isWritable() && !emptyQueue) { StoredMessage msg = clientSession.queue().poll(); if (msg == null) { emptyQueue = true; } else { - //recreate a publish from stored publish in queue + // 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); } }