diff --git a/broker/src/main/java/io/moquette/spi/impl/MetaPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MetaPublisher.java new file mode 100644 index 00000000..26397ae7 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/MetaPublisher.java @@ -0,0 +1,78 @@ +package io.moquette.spi.impl; + +import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.parser.proto.messages.PublishMessage; +import io.moquette.server.ConnectionDescriptor; +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.netty.channel.Channel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.ConcurrentMap; + +import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired; + +class MetaPublisher { + + private static final Logger LOG = LoggerFactory.getLogger(MetaPublisher.class); + private final ConcurrentMap connectionDescriptors; + private final ISessionsStore m_sessionsStore; + private final IMessagesStore m_messagesStore; + private final Qos0Publisher qos0Publisher; + private final Qos2Publisher qos2Publisher; + + public MetaPublisher(ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, + IMessagesStore messagesStore) { + this.connectionDescriptors = connectionDescriptors; + this.m_sessionsStore = sessionsStore; + this.m_messagesStore = messagesStore; + this.qos0Publisher = new Qos0Publisher(connectionDescriptors, sessionsStore); + this.qos2Publisher = new Qos2Publisher(connectionDescriptors, sessionsStore, m_messagesStore); + } + + void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { + final String topic = pubMsg.getTopic(); + final AbstractMessage.QOSType publishingQos = pubMsg.getQos(); + final ByteBuffer origMessage = pubMsg.getMessage(); + + //if QoS 1 or 2 store the message + MessageGUID guid = null; + 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.containsKey(sub.getClientId()); + + LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", + sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); + ByteBuffer message = origMessage.duplicate(); + if (targetIsActive) { + //QoS 0 + if (qos == AbstractMessage.QOSType.MOST_ONE) { + this.qos0Publisher.publishQos0(targetSession, topic, message); + } else { + //QoS 1 or 2 + int messageId = targetSession.nextPacketId(); + targetSession.inFlightAckWaiting(guid, messageId); + this.qos2Publisher.publishQos2(targetSession, topic, qos, message, false, messageId); + } + } else { + if (!targetSession.isCleanSession()) { + //store the message in targetSession queue to deliver + targetSession.enqueueToDeliver(guid); + } + } + } + } +} 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 129bedc5..4db2395c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -136,7 +136,7 @@ public class ProtocolProcessor { private Qos0PublishHandler qos0PublishHandler; private Qos1PublishHandler qos1PublishHandler; private Qos2PublishHandler qos2PublishHandler; - private Qos2Publisher qos2Publisher; + private MetaPublisher metaPublisher; private InternalRepublisher internalRepublisher; //maps clientID to Will testament, if specified on CONNECT @@ -194,9 +194,9 @@ public class ProtocolProcessor { this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port); - this.qos2Publisher = new Qos2Publisher(connectionDescriptors, sessionsStore, m_messagesStore); + this.metaPublisher = new MetaPublisher(connectionDescriptors, sessionsStore, m_messagesStore); this.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, - m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port, this.qos2Publisher); + m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port/*, this.qos2Publisher*/); this.internalRepublisher = new InternalRepublisher(this.connectionDescriptors); } @@ -488,7 +488,7 @@ public class ProtocolProcessor { guid = m_messagesStore.storePublishForFuture(toStoreMsg); } List topicMatchingSubscriptions = subscriptions.matches(topic); - this.qos2Publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); + this.metaPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); if (!msg.isRetainFlag()) { return; @@ -522,7 +522,7 @@ public class ProtocolProcessor { String topic = tobeStored.getTopic(); List topicMatchingSubscriptions = subscriptions.matches(topic); - this.qos2Publisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions); + this.metaPublisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions); } static QOSType lowerQosToTheSubscriptionDesired(Subscription sub, QOSType qos) { 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 925c3fd4..f8586f1c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -25,7 +25,7 @@ class Qos0PublishHandler { private final SubscriptionsStore subscriptions; private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; - private final Qos0Publisher qos0Publisher; + private final MetaPublisher metaPublisher; public Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap topicMatchingSubscriptions = subscriptions.matches(topic); - qos0Publisher.publish2Subscribers_qos0(toStoreMsg, topicMatchingSubscriptions); + this.metaPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); if (msg.isRetainFlag()) { //QoS == 0 && retain => clean old retained diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos0Publisher.java b/broker/src/main/java/io/moquette/spi/impl/Qos0Publisher.java index 5c256279..7c320d4b 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0Publisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0Publisher.java @@ -26,30 +26,7 @@ class Qos0Publisher { this.m_sessionsStore = sessionsStore; } - void publish2Subscribers_qos0(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { - final String topic = pubMsg.getTopic(); - final ByteBuffer origMessage = pubMsg.getMessage(); -// LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic); -// if (LOG.isTraceEnabled()) { -// LOG.trace("content <{}>", DebugUtils.payload2Str(origMessage)); -// LOG.trace("subscription tree {}", subscriptions.dumpTree()); -// } - - LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic); - for (final Subscription sub : topicMatchingSubscriptions) { - boolean targetIsActive = connectionDescriptors.containsKey(sub.getClientId()); - - LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", - sub.getClientId(), sub.getTopicFilter(), AbstractMessage.QOSType.MOST_ONE, targetIsActive); - if (targetIsActive) { - ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); - ByteBuffer message = origMessage.duplicate(); - publishQos0(targetSession, topic, message); - } - } - } - - private void publishQos0(ClientSession clientsession, String topic, ByteBuffer message) { + void publishQos0(ClientSession clientsession, String topic, ByteBuffer message) { String clientId = clientsession.clientID; LOG.debug("publishQos0 invoked clientId <{}> on topic <{}>", clientId, topic); 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 3caa4da3..400201c9 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -28,7 +28,7 @@ class Qos1PublishHandler { private final BrokerInterceptor m_interceptor; private final ConcurrentMap connectionDescriptors; private final String brokerPort; - private final Qos1Publisher qos1Publisher; + private final MetaPublisher metaPublisher; public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap topicMatchingSubscriptions = subscriptions.matches(topic); - this.qos1Publisher.publish2Subscribers_qos1(toStoreMsg, topicMatchingSubscriptions); + this.metaPublisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); //send PUBACK final Integer messageID = msg.getMessageID(); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java b/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java deleted file mode 100644 index e0a9c07f..00000000 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java +++ /dev/null @@ -1,121 +0,0 @@ -package io.moquette.spi.impl; - -import io.moquette.parser.proto.messages.AbstractMessage; -import io.moquette.parser.proto.messages.PublishMessage; -import io.moquette.server.ConnectionDescriptor; -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.netty.channel.Channel; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.nio.ByteBuffer; -import java.util.List; -import java.util.concurrent.ConcurrentMap; - -import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired; - -class Qos1Publisher { - - private static final Logger LOG = LoggerFactory.getLogger(Qos1Publisher.class); - private final ConcurrentMap connectionDescriptors; - private final ISessionsStore m_sessionsStore; - private final IMessagesStore m_messagesStore; - - public Qos1Publisher(ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, - IMessagesStore messagesStore) { - this.connectionDescriptors = connectionDescriptors; - this.m_sessionsStore = sessionsStore; - m_messagesStore = messagesStore; - } - - void publish2Subscribers_qos1(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { - final String topic = pubMsg.getTopic(); - final AbstractMessage.QOSType publishingQos = pubMsg.getQos(); - final ByteBuffer origMessage = pubMsg.getMessage(); - - //if QoS 1 or 2 store the message - MessageGUID 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.containsKey(sub.getClientId()); - - LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", - sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); - ByteBuffer message = origMessage.duplicate(); - if (qos == AbstractMessage.QOSType.MOST_ONE && targetIsActive) { - //QoS 0 - publishQos1(targetSession, topic, qos, message, false, null); - } else { - //QoS 1 or 2 - //if the target subscription is not clean session and is not connected => store it - if (!targetSession.isCleanSession() && !targetIsActive) { - //store the message in targetSession queue to deliver - targetSession.enqueueToDeliver(guid); - } else { - //publish - if (targetIsActive) { - int messageId = targetSession.nextPacketId(); - targetSession.inFlightAckWaiting(guid, messageId); - publishQos1(targetSession, topic, qos, message, false, messageId); - } - } - } - } - } - - private void publishQos1(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, - ByteBuffer message, boolean retained, Integer messageID) { - String clientId = clientsession.clientID; - LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", - clientId, topic, qos, retained, messageID); - PublishMessage pubMessage = new PublishMessage(); - pubMessage.setRetainFlag(retained); - pubMessage.setTopicName(topic); - pubMessage.setQos(qos); - pubMessage.setPayload(message); - - LOG.info("send publish message to <{}> on topic <{}>", clientId, topic); - if (LOG.isDebugEnabled()) { - LOG.debug("content <{}>", DebugUtils.payload2Str(message)); - } - //set the PacketIdentifier only for QoS > 0 - if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) { - pubMessage.setMessageID(messageID); - } else { - if (messageID != null) { - throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message " + - "with PacketIdentifier: " + messageID); - } - } - - if (connectionDescriptors == null) { - throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " + - "initialized, somewhere it's overwritten!!"); - } - if (connectionDescriptors.get(clientId) == null) { - //TODO while we were publishing to the target client, that client disconnected, - // could happen is not an error HANDLE IT - throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", - clientId, connectionDescriptors)); - } - Channel channel = connectionDescriptors.get(clientId).channel; - LOG.trace("Session for clientId {}", clientId); - if (channel.isWritable()) { - LOG.debug("channel is writable"); - //if channel is writable don't enqueue - channel.writeAndFlush(pubMessage); - } else { - //enqueue to the client session - LOG.debug("enqueue to client session"); - clientsession.enqueue(pubMessage); - } - } -} 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 ffa6eb03..075411aa 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -30,12 +30,12 @@ class Qos2PublishHandler { private final ConcurrentMap connectionDescriptors; private final ISessionsStore m_sessionsStore; private final String brokerPort; - private final Qos2Publisher qos2Publisher; + private final MetaPublisher metaPublisher; public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, String brokerPort, - Qos2Publisher qos2Publisher) { + ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore, String brokerPort/*, + Qos2Publisher qos2Publisher*/) { this.m_authorizator = authorizator; this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; @@ -43,7 +43,7 @@ class Qos2PublishHandler { this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; this.brokerPort = brokerPort; - this.qos2Publisher = qos2Publisher; + this.metaPublisher = new MetaPublisher(connectionDescriptors, sessionsStore, messagesStore); } void receivedPublishQos2(Channel channel, PublishMessage msg) { @@ -95,7 +95,7 @@ class Qos2PublishHandler { LOG.trace("content <{}>", DebugUtils.payload2Str(evt.getMessage())); LOG.trace("subscription tree {}", subscriptions.dumpTree()); } - this.qos2Publisher.publish2Subscribers(evt, topicMatchingSubscriptions); + this.metaPublisher.publish2Subscribers(evt, topicMatchingSubscriptions); if (evt.isRetained()) { if (!evt.getMessage().hasRemaining()) { diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2Publisher.java b/broker/src/main/java/io/moquette/spi/impl/Qos2Publisher.java index 3f210641..0bc2a188 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2Publisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2Publisher.java @@ -6,18 +6,13 @@ import io.moquette.server.ConnectionDescriptor; 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.netty.channel.Channel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.nio.ByteBuffer; -import java.util.List; import java.util.concurrent.ConcurrentMap; -import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired; - class Qos2Publisher { private static final Logger LOG = LoggerFactory.getLogger(Qos2Publisher.class); @@ -32,48 +27,7 @@ class Qos2Publisher { m_messagesStore = messagesStore; } - void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { - final String topic = pubMsg.getTopic(); - final AbstractMessage.QOSType publishingQos = pubMsg.getQos(); - final ByteBuffer origMessage = pubMsg.getMessage(); - - //if QoS 1 or 2 store the message - MessageGUID guid = null; - 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.containsKey(sub.getClientId()); - - LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", - sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive); - ByteBuffer message = origMessage.duplicate(); - if (targetIsActive) { - //QoS 0 - if (qos == AbstractMessage.QOSType.MOST_ONE) { - publishQos2(targetSession, topic, qos, message, false, null); - } else { - //QoS 1 or 2 - int messageId = targetSession.nextPacketId(); - targetSession.inFlightAckWaiting(guid, messageId); - publishQos2(targetSession, topic, qos, message, false, messageId); - } - } else { - if (!targetSession.isCleanSession()) { - //store the message in targetSession queue to deliver - targetSession.enqueueToDeliver(guid); - } - } - } - } - - - private void publishQos2(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, + void publishQos2(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) { String clientId = clientsession.clientID; LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}",