From a9f53a4a956db486e5b103598f5e29006683788a Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 11 Dec 2016 10:45:36 +0100 Subject: [PATCH] Moved Qos2 publish handling code to separate class --- .../moquette/spi/impl/ProtocolProcessor.java | 148 ++++++------ .../moquette/spi/impl/Qos2PublishHandler.java | 223 ++++++++++++++++++ 2 files changed, 299 insertions(+), 72 deletions(-) create mode 100644 broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java 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 39611252..bda544b8 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -135,6 +135,7 @@ public class ProtocolProcessor { private Qos0PublishHandler qos0PublishHandler; private Qos1PublishHandler qos1PublishHandler; + private Qos2PublishHandler qos2PublishHandler; //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); @@ -190,6 +191,8 @@ 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.qos2PublishHandler = new Qos2PublishHandler(m_authorizator, subscriptions, m_messagesStore, + m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port); } public void processConnect(Channel channel, ConnectMessage msg) { @@ -453,52 +456,52 @@ public class ProtocolProcessor { this.qos1PublishHandler.receivedPublishQos1(channel, msg); break; case EXACTLY_ONCE: - receivedPublishQos2(channel, msg); + this.qos2PublishHandler.receivedPublishQos2(channel, msg); break; } } - private void receivedPublishQos2(Channel channel, PublishMessage msg) { - final AbstractMessage.QOSType qos = QOSType.EXACTLY_ONCE; - final String topic = msg.getTopicName(); - //check if the topic can be wrote - if (checkWriteOnTopic(topic, channel)) { - return; - } - final Integer messageID = msg.getMessageID(); +// private void receivedPublishQos2(Channel channel, PublishMessage msg) { +// final AbstractMessage.QOSType qos = QOSType.EXACTLY_ONCE; +// final String topic = msg.getTopicName(); +// //check if the topic can be wrote +// if (checkWriteOnTopic(topic, channel)) { +// return; +// } +// final Integer messageID = msg.getMessageID(); +// +// IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); +// String clientID = NettyUtils.clientID(channel); +// toStoreMsg.setClientID(clientID); +// LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", m_server_port, clientID, topic, qos); +// //QoS2 +// MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); +// if (msg.isLocal()) { +// sendPubRec(clientID, messageID); +// } +// //Next the client will send us a pub rel +// //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher +// +// if (msg.isRetainFlag()) { +// if (!msg.getPayload().hasRemaining()) { +// m_messagesStore.cleanRetained(topic); +// } else { +// m_messagesStore.storeRetained(topic, guid); +// } +// } +// String username = NettyUtils.userName(channel); +// m_interceptor.notifyTopicPublished(msg, clientID, username); +// } - IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); - String clientID = NettyUtils.clientID(channel); - toStoreMsg.setClientID(clientID); - LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", m_server_port, clientID, topic, qos); - //QoS2 - MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); - if (msg.isLocal()) { - sendPubRec(clientID, messageID); - } - //Next the client will send us a pub rel - //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher - - if (msg.isRetainFlag()) { - if (!msg.getPayload().hasRemaining()) { - m_messagesStore.cleanRetained(topic); - } else { - m_messagesStore.storeRetained(topic, guid); - } - } - String username = NettyUtils.userName(channel); - m_interceptor.notifyTopicPublished(msg, clientID, username); - } - - boolean checkWriteOnTopic(String topic, Channel channel) { - String clientID = NettyUtils.clientID(channel); - String username = NettyUtils.userName(channel); - if (!m_authorizator.canWrite(topic, username, clientID)) { - LOG.debug("topic {} doesn't have write credentials", topic); - return true; - } - return false; - } +// boolean checkWriteOnTopic(String topic, Channel channel) { +// String clientID = NettyUtils.clientID(channel); +// String username = NettyUtils.userName(channel); +// if (!m_authorizator.canWrite(topic, username, clientID)) { +// LOG.debug("topic {} doesn't have write credentials", topic); +// return true; +// } +// return false; +// } /** * Intended usage is only for embedded versions of the broker, where the hosting application want to use the @@ -667,12 +670,12 @@ public class ProtocolProcessor { } } - private void sendPubRec(String clientID, int messageID) { - LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); - PubRecMessage pubRecMessage = new PubRecMessage(); - pubRecMessage.setMessageID(messageID); - connectionDescriptors.get(clientID).channel.writeAndFlush(pubRecMessage); - } +// private void sendPubRec(String clientID, int messageID) { +// LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); +// PubRecMessage pubRecMessage = new PubRecMessage(); +// pubRecMessage.setMessageID(messageID); +// connectionDescriptors.get(clientID).channel.writeAndFlush(pubRecMessage); +// } // private static void sendPubAck(String clientId, int messageID) { // LOG.trace("sendPubAck invoked"); @@ -698,33 +701,34 @@ public class ProtocolProcessor { * to all interested subscribers. * */ public void processPubRel(Channel channel, PubRelMessage msg) { - String clientID = NettyUtils.clientID(channel); - int messageID = msg.getMessageID(); - LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); - ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); - final String topic = evt.getTopic(); - List topicMatchingSubscriptions = subscriptions.matches(topic); - publish2Subscribers(evt, topicMatchingSubscriptions); - - if (evt.isRetained()) { - if (!evt.getMessage().hasRemaining()) { - m_messagesStore.cleanRetained(topic); - } else { - m_messagesStore.storeRetained(topic, evt.getGuid()); - } - } - - sendPubComp(clientID, messageID); + this.qos2PublishHandler.processPubRel(channel, msg); +// String clientID = NettyUtils.clientID(channel); +// int messageID = msg.getMessageID(); +// LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); +// ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); +// IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); +// final String topic = evt.getTopic(); +// List topicMatchingSubscriptions = subscriptions.matches(topic); +// publish2Subscribers(evt, topicMatchingSubscriptions); +// +// if (evt.isRetained()) { +// if (!evt.getMessage().hasRemaining()) { +// m_messagesStore.cleanRetained(topic); +// } else { +// m_messagesStore.storeRetained(topic, evt.getGuid()); +// } +// } +// +// sendPubComp(clientID, messageID); } - 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); - } +// 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); +// } public void processPubRec(Channel channel, PubRecMessage msg) { String clientID = NettyUtils.clientID(channel); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java new file mode 100644 index 00000000..2df29b2a --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java @@ -0,0 +1,223 @@ +package io.moquette.spi.impl; + +import io.moquette.parser.proto.messages.*; +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; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import io.moquette.spi.security.IAuthorizator; +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; + +public class Qos2PublishHandler { + + private static final Logger LOG = LoggerFactory.getLogger(Qos1PublishHandler.class); + + private final IAuthorizator m_authorizator; + private final SubscriptionsStore subscriptions; + private final IMessagesStore m_messagesStore; + private final BrokerInterceptor m_interceptor; + private final ConcurrentMap connectionDescriptors; + private final ISessionsStore m_sessionsStore; + private final String brokerPort; + + public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, + IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore, String brokerPort) { + this.m_authorizator = authorizator; + this.subscriptions = subscriptions; + this.m_messagesStore = messagesStore; + this.m_interceptor = interceptor; + this.connectionDescriptors = connectionDescriptors; + this.m_sessionsStore = sessionsStore; + this.brokerPort = brokerPort; + } + + void receivedPublishQos2(Channel channel, PublishMessage msg) { + final AbstractMessage.QOSType qos = AbstractMessage.QOSType.EXACTLY_ONCE; + final String topic = msg.getTopicName(); + //check if the topic can be wrote + if (checkWriteOnTopic(topic, channel)) { + return; + } + final Integer messageID = msg.getMessageID(); + + IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); + String clientID = NettyUtils.clientID(channel); + toStoreMsg.setClientID(clientID); + LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", this.brokerPort, clientID, topic, qos); + //QoS2 + MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); + if (msg.isLocal()) { + sendPubRec(clientID, messageID); + } + //Next the client will send us a pub rel + //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher + + if (msg.isRetainFlag()) { + if (!msg.getPayload().hasRemaining()) { + m_messagesStore.cleanRetained(topic); + } else { + m_messagesStore.storeRetained(topic, guid); + } + } + String username = NettyUtils.userName(channel); + m_interceptor.notifyTopicPublished(msg, clientID, username); + } + + /** + * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish + * to all interested subscribers. + * */ + void processPubRel(Channel channel, PubRelMessage msg) { + String clientID = NettyUtils.clientID(channel); + int messageID = msg.getMessageID(); + LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); + ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); + IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); + final String topic = evt.getTopic(); + List topicMatchingSubscriptions = subscriptions.matches(topic); + publish2Subscribers(evt, topicMatchingSubscriptions); + + if (evt.isRetained()) { + if (!evt.getMessage().hasRemaining()) { + m_messagesStore.cleanRetained(topic); + } else { + m_messagesStore.storeRetained(topic, evt.getGuid()); + } + } + + sendPubComp(clientID, messageID); + } + + private boolean checkWriteOnTopic(String topic, Channel channel) { + String clientID = NettyUtils.clientID(channel); + String username = NettyUtils.userName(channel); + if (!m_authorizator.canWrite(topic, username, clientID)) { + LOG.debug("topic {} doesn't have write credentials", topic); + return true; + } + return false; + } + + private void sendPubRec(String clientID, int messageID) { + LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); + PubRecMessage pubRecMessage = new PubRecMessage(); + pubRecMessage.setMessageID(messageID); + 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); + } +}