From 5b254cfe75d09c10bfe8a23ab8e3c73e2ae05d42 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 11 Dec 2016 18:50:10 +0100 Subject: [PATCH] Changed che willpublish to use the Qos2Publisher helper class --- .../moquette/spi/impl/ProtocolProcessor.java | 62 +++---------------- .../moquette/spi/impl/Qos2PublishHandler.java | 5 +- 2 files changed, 11 insertions(+), 56 deletions(-) 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 1a0e652a..de7bf5f8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -136,6 +136,7 @@ public class ProtocolProcessor { private Qos0PublishHandler qos0PublishHandler; private Qos1PublishHandler qos1PublishHandler; private Qos2PublishHandler qos2PublishHandler; + private Qos2Publisher qos2Publisher; //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); @@ -191,8 +192,10 @@ public class ProtocolProcessor { m_interceptor, this.connectionDescriptors, m_sessionsStore); 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.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, - m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port); + m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port, this.qos2Publisher); } public void processConnect(Channel channel, ConnectMessage msg) { @@ -486,7 +489,8 @@ public class ProtocolProcessor { guid = m_messagesStore.storePublishForFuture(toStoreMsg); } List topicMatchingSubscriptions = subscriptions.matches(topic); - publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); +// publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); + this.qos2Publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); if (!msg.isRetainFlag()) { return; @@ -519,60 +523,10 @@ public class ProtocolProcessor { tobeStored.setMessageID(messageId); String topic = tobeStored.getTopic(); List topicMatchingSubscriptions = subscriptions.matches(topic); - publish2Subscribers(tobeStored, topicMatchingSubscriptions); + + this.qos2Publisher.publish2Subscribers(tobeStored, topicMatchingSubscriptions); } - - /** - * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 - * */ - 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 - directSend(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); - directSend(targetSession, topic, qos, message, false, messageId); - } - } - } - } - } - - static QOSType lowerQosToTheSubscriptionDesired(Subscription sub, QOSType qos) { if (qos.byteValue() > sub.getRequestedQos().byteValue()) { qos = sub.getRequestedQos(); 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 b5dc0b84..ffa6eb03 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -34,7 +34,8 @@ class Qos2PublishHandler { public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, String brokerPort) { + ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore, String brokerPort, + Qos2Publisher qos2Publisher) { this.m_authorizator = authorizator; this.subscriptions = subscriptions; this.m_messagesStore = messagesStore; @@ -42,7 +43,7 @@ class Qos2PublishHandler { this.connectionDescriptors = connectionDescriptors; this.m_sessionsStore = sessionsStore; this.brokerPort = brokerPort; - this.qos2Publisher = new Qos2Publisher(connectionDescriptors, sessionsStore, messagesStore); + this.qos2Publisher = qos2Publisher; } void receivedPublishQos2(Channel channel, PublishMessage msg) {