diff --git a/broker/src/main/java/io/moquette/server/HazelcastListener.java b/broker/src/main/java/io/moquette/server/HazelcastListener.java index 1399e147..0fc943fc 100644 --- a/broker/src/main/java/io/moquette/server/HazelcastListener.java +++ b/broker/src/main/java/io/moquette/server/HazelcastListener.java @@ -40,11 +40,8 @@ public class HazelcastListener implements MessageListener { try { if (!msg.getPublishingMember().equals(server.getHazelcastInstance().getCluster().getLocalMember())) { HazelcastMsg hzMsg = msg.getMessageObject(); - LOG.info( - "{} received from hazelcast for topic {} message: {}", - hzMsg.getClientId(), - hzMsg.getTopic(), - hzMsg.getPayload()); + LOG.info("{} received from hazelcast for topic {} message: {}", hzMsg.getClientId(), hzMsg.getTopic(), + hzMsg.getPayload()); // TODO pass forward this information in somehow publishMessage.setLocal(false); MqttQoS qos = MqttQoS.valueOf(hzMsg.getQos()); diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 483123fe..d31d5c2d 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -198,8 +198,15 @@ public class ClientSession { * * @return the packetID for the message in flight. * */ - public int inFlightAckWaiting(MessageGUID guid) { - LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}", clientID, guid); + public int inFlightAckWaiting(IMessagesStore.StoredMessage msg) { + LOG.debug("Adding message ot inflight zone. ClientId={}", clientID); + MessageGUID guid = msg.getGuid(); + if (guid == null) { + guid = messagesStore.storePublishForFuture(msg); + msg.setGuid(guid); + LOG.debug("Not yet assigned guid to message, updating guid={}", guid); + } + int messageId = ClientSession.this.nextPacketId(); outboundFlightZone.waitingAck(messageId, guid); return messageId; diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 9b853a50..beb6a373 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -49,10 +49,6 @@ public interface IMessagesStore { return m_qos; } - public ByteBuffer getPayload() { - return (ByteBuffer) ByteBuffer.allocate(m_payload.length).put(m_payload).flip(); - } - public String getTopic() { return m_topic; } @@ -73,7 +69,7 @@ public interface IMessagesStore { this.m_clientID = m_clientID; } - public ByteBuf getMessage() { + public ByteBuf getPayload() { return Unpooled.copiedBuffer(m_payload); } diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java index 77e3552d..7449ae74 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -44,7 +44,7 @@ class InternalRepublisher { if (storedMsg.getQos() != MqttQoS.AT_MOST_ONCE) { LOG.debug("Adding message to inflight zone. ClientId={}, topic={}", targetSession.clientID, storedMsg.getTopic()); - int packetID = targetSession.inFlightAckWaiting(storedMsg.getGuid()); + int packetID = targetSession.inFlightAckWaiting(storedMsg); // set the PacketIdentifier only for QoS > 0 publishMsg = retainedPublish(storedMsg, packetID); @@ -62,7 +62,7 @@ class InternalRepublisher { // put in flight zone LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}, topic={}", clientSession.clientID, pubEvt.getGuid(), pubEvt.getTopic()); - int messageId = clientSession.inFlightAckWaiting(pubEvt.getGuid()); + int messageId = clientSession.inFlightAckWaiting(pubEvt); MqttPublishMessage publishMsg = notRetainedPublish(pubEvt); // set the PacketIdentifier only for QoS > 0 if (publishMsg.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) { @@ -73,22 +73,22 @@ class InternalRepublisher { } private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage, Integer messageID) { - return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false, + return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), false, messageID); } private MqttPublishMessage notRetainedPublish(IMessagesStore.StoredMessage storedMessage) { - return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), false, + return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), false, 0); } private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage) { - return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true, + return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), true, 0); } private MqttPublishMessage retainedPublish(IMessagesStore.StoredMessage storedMessage, Integer packetID) { - return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true, + return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getPayload(), true, packetID); } diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index 9fd9e518..7d833078 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java @@ -20,8 +20,9 @@ import io.moquette.server.ConnectionDescriptorStore; import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.impl.subscriptions.Topic; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; @@ -34,15 +35,15 @@ class MessagesPublisher { private static final Logger LOG = LoggerFactory.getLogger(MessagesPublisher.class); private final ConnectionDescriptorStore connectionDescriptors; private final ISessionsStore m_sessionsStore; - private final IMessagesStore m_messagesStore; private final PersistentQueueMessageSender messageSender; + private final SubscriptionsStore subscriptions; - MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, - IMessagesStore messagesStore, PersistentQueueMessageSender messageSender) { + public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, + PersistentQueueMessageSender messageSender, SubscriptionsStore subscriptions) { this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; - this.m_messagesStore = messagesStore; this.messageSender = messageSender; + this.subscriptions = subscriptions; } static MqttPublishMessage notRetainedPublish(String topic, MqttQoS qos, ByteBuf message) { @@ -56,36 +57,44 @@ class MessagesPublisher { return new MqttPublishMessage(fixedHeader, varHeader, message); } - void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { - final String topic = pubMsg.getTopic(); - final MqttQoS publishingQos = pubMsg.getQos(); - final ByteBuf origMessage = pubMsg.getMessage(); - - // if QoS 1 or 2 store the message - // TODO perhaps this block is not needed. - MessageGUID guid = null; - if (publishingQos != MqttQoS.AT_MOST_ONCE) { - guid = m_messagesStore.storePublishForFuture(pubMsg); + void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic, int messageID) { + if (LOG.isTraceEnabled()) { + LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + + "subscriptionTree={}", pubMsg.getClientID(), topic, messageID, DebugUtils.payload2Str(pubMsg.getPayload()), + subscriptions.dumpTree()); + } else { + LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", pubMsg.getClientID(), topic, + messageID); } + publish2Subscribers(pubMsg, topic); + } + + void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, Topic topic) { + List topicMatchingSubscriptions = subscriptions.matches(topic); + final String topic1 = pubMsg.getTopic(); + final MqttQoS publishingQos = pubMsg.getQos(); + final ByteBuf origPayload = pubMsg.getPayload(); for (final Subscription sub : topicMatchingSubscriptions) { MqttQoS qos = lowerQosToTheSubscriptionDesired(sub, publishingQos); ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); boolean targetIsActive = this.connectionDescriptors.isConnected(sub.getClientId()); - - // we need to retain because duplicate only copy r/w indexes and don't retain() causing - // refCnt = 0 - ByteBuf message = origMessage.retainedDuplicate(); +//TODO move all this logic into messageSender, which puts into the flightZone only the messages that pull out of the queue. if (targetIsActive) { LOG.debug("Sending PUBLISH message to active subscriber. CId={}, topicFilter={}, qos={}", sub.getClientId(), sub.getTopicFilter(), qos); - MqttPublishMessage publishMsg = notRetainedPublish(topic, qos, message); + // we need to retain because duplicate only copy r/w indexes and don't retain() causing + // refCnt = 0 + ByteBuf payload = origPayload.retainedDuplicate(); + MqttPublishMessage publishMsg; if (qos != MqttQoS.AT_MOST_ONCE) { // QoS 1 or 2 - int messageId = targetSession.inFlightAckWaiting(guid); + int messageId = targetSession.inFlightAckWaiting(pubMsg); // set the PacketIdentifier only for QoS > 0 - publishMsg = notRetainedPublishWithMessageId(topic, qos, message, messageId); + publishMsg = notRetainedPublishWithMessageId(topic1, qos, payload, messageId); + } else { + publishMsg = notRetainedPublish(topic1, qos, payload); } this.messageSender.sendPublish(targetSession, publishMsg); } else { @@ -98,4 +107,5 @@ class MessagesPublisher { } } } + } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 96791fce..6ed4de8b 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -216,16 +216,16 @@ 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, messageSender, + subscriptions); LOG.info("Initializing QoS publish handlers..."); - this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor, + this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, 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.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, m_messagesStore, m_interceptor, + this.connectionDescriptors, this.messagesPublisher); this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor, - this.connectionDescriptors, m_sessionsStore, m_server_port, this.messagesPublisher); + this.connectionDescriptors, m_sessionsStore, this.messagesPublisher); LOG.info("Initializing internal republisher..."); this.internalRepublisher = new InternalRepublisher(messageSender); @@ -542,8 +542,7 @@ public class ProtocolProcessor { if (qos == EXACTLY_ONCE) { // QoS2 guid = m_messagesStore.storePublishForFuture(toStoreMsg); } - List topicMatchingSubscriptions = subscriptions.matches(topic); - this.messagesPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); + this.messagesPublisher.publish2Subscribers(toStoreMsg, topic); if (!msg.fixedHeader().isRetain()) { return; @@ -564,15 +563,13 @@ public class ProtocolProcessor { * Specialized version to publish will testament message. */ private void forwardPublishWill(WillMessage will, String clientID) { + LOG.info("Publishing will message. CId={}, topic={}", clientID, will.getTopic()); // 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 IMessagesStore.StoredMessage tobeStored = asStoredMessage(will); tobeStored.setClientID(clientID); Topic topic = new Topic(tobeStored.getTopic()); - List topicMatchingSubscriptions = subscriptions.matches(topic); - - LOG.info("Publishing will message. CId={}, topic={}", clientID, topic); - this.messagesPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions); + this.messagesPublisher.publish2Subscribers(tobeStored, topic); } static MqttQoS lowerQosToTheSubscriptionDesired(Subscription sub, MqttQoS qos) { @@ -922,7 +919,7 @@ 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(), + MqttPublishMessage pubMsg = createPublishForQos( msg.getTopic(), msg.getQos(), msg.getPayload(), retained, 0); channel.write(pubMsg); } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java index 9f730511..e48511ef 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -18,30 +18,25 @@ package io.moquette.spi.impl; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; -import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.moquette.spi.impl.subscriptions.Topic; import io.moquette.spi.security.IAuthorizator; import io.netty.channel.Channel; import io.netty.handler.codec.mqtt.MqttPublishMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.List; import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage; class Qos0PublishHandler extends QosPublishHandler { private static final Logger LOG = LoggerFactory.getLogger(Qos0PublishHandler.class); - private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; private final MessagesPublisher publisher; - Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, - BrokerInterceptor interceptor, MessagesPublisher messagesPublisher) { + public Qos0PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor, + MessagesPublisher messagesPublisher) { super(authorizator); - this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; this.publisher = messagesPublisher; @@ -50,35 +45,24 @@ class Qos0PublishHandler extends QosPublishHandler { void receivedPublishQos0(Channel channel, MqttPublishMessage msg) { // verify if topic can be write final Topic topic = new Topic(msg.variableHeader().topicName()); - if (checkWriteOnTopic(topic, channel)) { + String clientID = NettyUtils.clientID(channel); + String username = NettyUtils.userName(channel); + if (!m_authorizator.canWrite(topic, username, clientID)) { + LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic); return; } // route message to subscribers IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); - String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - final int messageID = msg.variableHeader().messageId(); - - if (LOG.isTraceEnabled()) { - LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + - "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()), - subscriptions.dumpTree()); - } else { - LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic, - messageID); - } - - List topicMatchingSubscriptions = subscriptions.matches(topic); - this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); + this.publisher.publish2Subscribers(toStoreMsg, topic); if (msg.fixedHeader().isRetain()) { // QoS == 0 && retain => clean old retained m_messagesStore.cleanRetained(topic); } - String username = NettyUtils.userName(channel); m_interceptor.notifyTopicPublished(msg, clientID, username); } } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 378046ee..b44edf86 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -39,17 +39,14 @@ class Qos1PublishHandler extends QosPublishHandler { private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); - private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; private final ConnectionDescriptorStore connectionDescriptors; private final MessagesPublisher publisher; - Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, - BrokerInterceptor interceptor, ConnectionDescriptorStore connectionDescriptors, String brokerPort, - MessagesPublisher messagesPublisher) { + public Qos1PublishHandler(IAuthorizator authorizator, IMessagesStore messagesStore, BrokerInterceptor interceptor, + ConnectionDescriptorStore connectionDescriptors, MessagesPublisher messagesPublisher) { super(authorizator); - this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; this.m_interceptor = interceptor; this.connectionDescriptors = connectionDescriptors; @@ -59,7 +56,10 @@ class Qos1PublishHandler extends QosPublishHandler { void receivedPublishQos1(Channel channel, MqttPublishMessage msg) { // verify if topic can be write final Topic topic = new Topic(msg.variableHeader().topicName()); - if (checkWriteOnTopic(topic, channel)) { + String clientID = NettyUtils.clientID(channel); + String username = NettyUtils.userName(channel); + if (!m_authorizator.canWrite(topic, username, clientID)) { + LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic); return; } @@ -67,20 +67,9 @@ class Qos1PublishHandler extends QosPublishHandler { // route message to subscribers IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); - String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); - if (LOG.isTraceEnabled()) { - LOG.trace("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}, payload={}, " + - "subscriptionTree={}", clientID, topic, messageID, DebugUtils.payload2Str(toStoreMsg.getMessage()), - subscriptions.dumpTree()); - } else { - LOG.info("Sending publish message to subscribers. ClientId={}, topic={}, messageId={}", clientID, topic, - messageID); - } - - List topicMatchingSubscriptions = subscriptions.matches(topic); - this.publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); + this.publisher.publish2Subscribers(toStoreMsg, topic, messageID); sendPubAck(clientID, messageID); @@ -94,7 +83,6 @@ class Qos1PublishHandler extends QosPublishHandler { } } - String username = NettyUtils.userName(channel); m_interceptor.notifyTopicPublished(msg, clientID, username); } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index 447adfab..37021b22 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -49,10 +49,10 @@ class Qos2PublishHandler extends QosPublishHandler { private final ISessionsStore m_sessionsStore; private final MessagesPublisher publisher; - Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, - IMessagesStore messagesStore, BrokerInterceptor interceptor, - ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, String brokerPort, - MessagesPublisher messagesPublisher) { + public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, + IMessagesStore messagesStore, BrokerInterceptor interceptor, + ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore, + MessagesPublisher messagesPublisher) { super(authorizator); this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; @@ -65,18 +65,20 @@ class Qos2PublishHandler extends QosPublishHandler { void receivedPublishQos2(Channel channel, MqttPublishMessage msg) { final Topic topic = new Topic(msg.variableHeader().topicName()); // check if the topic can be wrote - if (checkWriteOnTopic(topic, channel)) { + String clientID = NettyUtils.clientID(channel); + String username = NettyUtils.userName(channel); + if (!m_authorizator.canWrite(topic, username, clientID)) { + LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic); return; } final int messageID = msg.variableHeader().messageId(); IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); - String clientID = NettyUtils.clientID(channel); toStoreMsg.setClientID(clientID); LOG.info("Sending publish message to subscribers CId={}, topic={}, messageId={}", clientID, topic, messageID); if (LOG.isTraceEnabled()) { - LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getMessage()), subscriptions.dumpTree()); + LOG.trace("payload={}, subs Tree={}", payload2Str(toStoreMsg.getPayload()), subscriptions.dumpTree()); } // QoS2 @@ -96,7 +98,6 @@ class Qos2PublishHandler extends QosPublishHandler { m_messagesStore.storeRetained(topic, guid); } } - String username = NettyUtils.userName(channel); m_interceptor.notifyTopicPublished(msg, clientID, username); } @@ -115,20 +116,11 @@ class Qos2PublishHandler extends QosPublishHandler { throw new IllegalArgumentException("Can't find inbound inflight message"); } final Topic topic = new Topic(evt.getTopic()); - List topicMatchingSubscriptions = subscriptions.matches(topic); - if (LOG.isDebugEnabled()) { - LOG.debug("Sending publish message to subscribers. CId={}, topic={}, messageId={}, payload={}, " + - "subscriptionTree={}", clientID, topic, messageID, payload2Str(evt.getMessage()), - subscriptions.dumpTree()); - } else { - LOG.info("Sending publish message to subscribers. CId={}, topic={}, messageId={}", clientID, topic, - messageID); - } - this.publisher.publish2Subscribers(evt, topicMatchingSubscriptions); + this.publisher.publish2Subscribers(evt, topic, messageID); if (evt.isRetained()) { - if (evt.getMessage().readableBytes() == 0) { + if (evt.getPayload().readableBytes() == 0) { m_messagesStore.cleanRetained(topic); } else { m_messagesStore.storeRetained(topic, evt.getGuid()); diff --git a/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java index e318aecf..5a85bc8c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/QosPublishHandler.java @@ -27,7 +27,7 @@ abstract class QosPublishHandler { private static final Logger LOG = LoggerFactory.getLogger(QosPublishHandler.class); - private final IAuthorizator m_authorizator; + protected final IAuthorizator m_authorizator; protected QosPublishHandler(IAuthorizator m_authorizator) { this.m_authorizator = m_authorizator; @@ -37,10 +37,7 @@ abstract class QosPublishHandler { 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); + LOG.error("MQTT client is not authorized to publish on topic. CId={}, topic={}", clientID, topic); return true; } return false; diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java index 8a74d138..51f8dd58 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java @@ -167,4 +167,11 @@ public class Topic implements Serializable { public int hashCode() { return topic.hashCode(); } + + /** + * Factory method + * */ + public static Topic asTopic(String s) { + return new Topic(s); + } }