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 3c939ec1..b5dc0b84 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -14,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 Qos2PublishHandler { @@ -32,6 +30,7 @@ class Qos2PublishHandler { private final ConcurrentMap connectionDescriptors; private final ISessionsStore m_sessionsStore; private final String brokerPort; + private final Qos2Publisher qos2Publisher; public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap topicMatchingSubscriptions = subscriptions.matches(topic); - publish2Subscribers(evt, topicMatchingSubscriptions); + 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()); + } + this.qos2Publisher.publish2Subscribers(evt, topicMatchingSubscriptions); if (evt.isRetained()) { if (!evt.getMessage().hasRemaining()) { @@ -119,105 +124,10 @@ class Qos2PublishHandler { connectionDescriptors.get(clientID).channel.writeAndFlush(pubRecMessage); } - void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { - final String topic = pubMsg.getTopic(); - final AbstractMessage.QOSType publishingQos = pubMsg.getQos(); - 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()); - } - //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 (qos == AbstractMessage.QOSType.MOST_ONE && targetIsActive) { - //QoS 0 - publishQos2(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); - publishQos2(targetSession, topic, qos, message, false, messageId); - } - } - } - } - } - - private 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 {}", - 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); - } - } - private void sendPubComp(String clientID, int messageID) { LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID); PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); - connectionDescriptors.get(clientID).channel.writeAndFlush(pubCompMessage); } } diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2Publisher.java b/broker/src/main/java/io/moquette/spi/impl/Qos2Publisher.java new file mode 100644 index 00000000..52d7907a --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2Publisher.java @@ -0,0 +1,126 @@ +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 Qos2Publisher { + + private static final Logger LOG = LoggerFactory.getLogger(Qos2Publisher.class); + private final ConcurrentMap connectionDescriptors; + private final ISessionsStore m_sessionsStore; + private final IMessagesStore m_messagesStore; + + public Qos2Publisher(ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, + IMessagesStore messagesStore) { + this.connectionDescriptors = connectionDescriptors; + this.m_sessionsStore = sessionsStore; + 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 (qos == AbstractMessage.QOSType.MOST_ONE && targetIsActive) { + //QoS 0 + publishQos2(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); + publishQos2(targetSession, topic, qos, message, false, messageId); + } + } + } + } + } + + + private 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 {}", + 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); + } + } + +}