anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

broker/ProtocolProcessor: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:39 +01:00
parent 6eb635f437
commit db31f77f9b
1 changed files with 324 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+324 -188
View File
@@ -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<String, WillMessage> 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<StoredMessage> 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<Subscription> 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<Subscription> 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<String> 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<MqttTopicSubscription> 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<Subscription> 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<Subscription> 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<MqttTopicSubscription> doVerify(String clientID, String username, MqttSubscribeMessage msg) {
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
List<MqttTopicSubscription> 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<MqttTopicSubscription> topicFilters, int messageId) {
List<Integer> 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<IMessagesStore.StoredMessage> 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);
}
}