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

Minor, just code layout

Browse Source
This commit is contained in:
andsel
2017-03-20 21:58:18 +01:00
parent 22b1fa2b0e
commit c659a54158
1 changed files with 94 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+94 -213
View File
@@ -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<StoredMessage> 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<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);
}
@@ -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<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 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<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("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<Subscription> 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);
}
}