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 4aef414c..5c256279 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0Publisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0Publisher.java @@ -7,7 +7,6 @@ import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.netty.channel.Channel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; 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 c2c84977..3caa4da3 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -1,11 +1,9 @@ package io.moquette.spi.impl; -import io.moquette.parser.proto.messages.AbstractMessage; import io.moquette.parser.proto.messages.PubAckMessage; import io.moquette.parser.proto.messages.PublishMessage; import io.moquette.server.ConnectionDescriptor; import io.moquette.server.netty.NettyUtils; -import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.MessageGUID; @@ -16,12 +14,10 @@ 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.asStoredMessage; -import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired; class Qos1PublishHandler { private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); @@ -31,8 +27,8 @@ class Qos1PublishHandler { private final IMessagesStore m_messagesStore; private final BrokerInterceptor m_interceptor; private final ConcurrentMap connectionDescriptors; - private final ISessionsStore m_sessionsStore; private final String brokerPort; + private final Qos1Publisher qos1Publisher; public Qos1PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap", DebugUtils.payload2Str(toStoreMsg.getMessage())); + LOG.trace("subscription tree {}", subscriptions.dumpTree()); + } List topicMatchingSubscriptions = subscriptions.matches(topic); - publish2Subscribers_qos1(toStoreMsg, topicMatchingSubscriptions); + this.qos1Publisher.publish2Subscribers_qos1(toStoreMsg, topicMatchingSubscriptions); //send PUBACK final Integer messageID = msg.getMessageID(); @@ -91,44 +93,6 @@ class Qos1PublishHandler { return false; } - void publish2Subscribers_qos1(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { - final String topic = pubMsg.getTopic(); - final AbstractMessage.QOSType publishingQos = pubMsg.getQos(); - final ByteBuffer origMessage = pubMsg.getMessage(); - LOG.debug("publish2Subscribers_qos1 republishing to existing subscribers that matches the topic {}", topic); - if (LOG.isTraceEnabled()) { - LOG.trace("content <{}>", DebugUtils.payload2Str(origMessage)); - LOG.trace("subscription tree {}", subscriptions.dumpTree()); - } - //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(); - //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, message, messageId, qos); - } - } - } - } - private void sendPubAck(String clientId, int messageID) { LOG.trace("sendPubAck invoked"); PubAckMessage pubAckMessage = new PubAckMessage(); @@ -148,44 +112,4 @@ class Qos1PublishHandler { } } - protected void publishQos1(ClientSession clientsession, String topic, - ByteBuffer message, int messageID, AbstractMessage.QOSType qos) { - String clientId = clientsession.clientID; - LOG.debug("publishQos1 invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", - clientId, topic, qos, false, messageID); - PublishMessage pubMessage = new PublishMessage(); - pubMessage.setRetainFlag(false); - pubMessage.setTopicName(topic); - pubMessage.setQos(qos); - pubMessage.setPayload(message); - //QoS > 0 => set the PacketIdentifier - pubMessage.setMessageID(messageID); - - LOG.info("send publish message to <{}> on topic <{}>", clientId, topic); - if (LOG.isDebugEnabled()) { - LOG.debug("content <{}>", DebugUtils.payload2Str(message)); - } - - 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/Qos1Publisher.java b/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java new file mode 100644 index 00000000..f1cc270f --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1Publisher.java @@ -0,0 +1,109 @@ +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(); + //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, message, messageId, qos); + } + } + } + } + + private void publishQos1(ClientSession clientsession, String topic, + ByteBuffer message, int messageID, AbstractMessage.QOSType qos) { + String clientId = clientsession.clientID; + LOG.debug("publishQos1 invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", + clientId, topic, qos, false, messageID); + PublishMessage pubMessage = new PublishMessage(); + pubMessage.setRetainFlag(false); + pubMessage.setTopicName(topic); + pubMessage.setQos(qos); + pubMessage.setPayload(message); + //QoS > 0 => set the PacketIdentifier + pubMessage.setMessageID(messageID); + + LOG.info("send publish message to <{}> on topic <{}>", clientId, topic); + if (LOG.isDebugEnabled()) { + LOG.debug("content <{}>", DebugUtils.payload2Str(message)); + } + + 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); + } + } +}