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

Improve logging. Remove duplicated code from QosPublishHandlers.

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