From 5d02239dc121479bf84e6d9d28985173f09ca36d Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 12 Dec 2016 22:30:14 +0100 Subject: [PATCH] Extracted common code for internal republish --- .../spi/impl/InternalRepublisher.java | 98 +++++++++++++++++++ .../moquette/spi/impl/ProtocolProcessor.java | 78 ++------------- 2 files changed, 105 insertions(+), 71 deletions(-) create mode 100644 broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java new file mode 100644 index 00000000..4a4384da --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -0,0 +1,98 @@ +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.netty.channel.Channel; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.ByteBuffer; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.ConcurrentMap; + +class InternalRepublisher { + + private static final Logger LOG = LoggerFactory.getLogger(InternalRepublisher.class); + + private final ConcurrentMap connectionDescriptors; + + InternalRepublisher(ConcurrentMap connectionDescriptors) { + this.connectionDescriptors = connectionDescriptors; + } + + void publishRetained(ClientSession targetSession, Collection messages) { + for (IMessagesStore.StoredMessage storedMsg : messages) { + //fire as retained the message + Integer packetID = storedMsg.getQos() == AbstractMessage.QOSType.MOST_ONE ? null : targetSession.nextPacketId(); + if (packetID != null) { + LOG.trace("Adding to inflight <{}>", packetID); + targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID); + } + directSend(targetSession, storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID); + } + } + + void publishStored(ClientSession clientSession, List publishedEvents) { + for (IMessagesStore.StoredMessage pubEvt : publishedEvents) { + //put in flight zone + LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID()); + clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID()); + directSend(clientSession, pubEvt.getTopic(), pubEvt.getQos(), + pubEvt.getMessage(), false, pubEvt.getMessageID()); + clientSession.removeEnqueued(pubEvt.getGuid()); + } + } + + private void directSend(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/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index de7bf5f8..be638741 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -137,6 +137,7 @@ public class ProtocolProcessor { private Qos1PublishHandler qos1PublishHandler; private Qos2PublishHandler qos2PublishHandler; private Qos2Publisher qos2Publisher; + private InternalRepublisher internalRepublisher; //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); @@ -196,6 +197,8 @@ public class ProtocolProcessor { 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, this.qos2Publisher); + + this.internalRepublisher = new InternalRepublisher(this.connectionDescriptors); } public void processConnect(Channel channel, ConnectMessage msg) { @@ -409,14 +412,7 @@ public class ProtocolProcessor { } LOG.info("republishing stored messages to client <{}>", clientSession.clientID); - for (IMessagesStore.StoredMessage pubEvt : publishedEvents) { - //put in flight zone - LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID()); - clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID()); - directSend(clientSession, pubEvt.getTopic(), pubEvt.getQos(), - pubEvt.getMessage(), false, pubEvt.getMessageID()); - clientSession.removeEnqueued(pubEvt.getGuid()); - } + this.internalRepublisher.publishStored(clientSession, publishedEvents); } public void processPubAck(Channel channel, PubAckMessage msg) { @@ -489,7 +485,6 @@ public class ProtocolProcessor { guid = m_messagesStore.storePublishForFuture(toStoreMsg); } List topicMatchingSubscriptions = subscriptions.matches(topic); -// publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); this.qos2Publisher.publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); if (!msg.isRetainFlag()) { @@ -534,54 +529,6 @@ public class ProtocolProcessor { return qos; } - protected void directSend(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); - } - } - /** * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish * to all interested subscribers. @@ -797,7 +744,7 @@ public class ProtocolProcessor { //fire the persisted messages in session for (Subscription subscription : newSubscriptions) { - publishStoredMessagesInSession(subscription, username); + publishRetainedMessagesInSession(subscription, username); } boolean success = this.subscriptionInCourse.remove(executionKey, SubscriptionState.STORED); @@ -860,9 +807,7 @@ public class ProtocolProcessor { return ackMessage; } - - - private void publishStoredMessagesInSession(final Subscription newSubscription, String username) { + private void publishRetainedMessagesInSession(final Subscription newSubscription, String username) { LOG.debug("Publish persisted messages in session {}", newSubscription); //scans retained messages to be published to the new subscription @@ -876,16 +821,7 @@ public class ProtocolProcessor { LOG.debug("Found {} messages to republish", messages.size()); ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId()); - for (IMessagesStore.StoredMessage storedMsg : messages) { - //fire as retained the message - LOG.trace("send publish message for topic {}", newSubscription.getTopicFilter()); - Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null : targetSession.nextPacketId(); - if (packetID != null) { - LOG.trace("Adding to inflight <{}>", packetID); - targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID); - } - directSend(targetSession, storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID); - } + this.internalRepublisher.publishRetained(targetSession, messages); //notify the Observables m_interceptor.notifyTopicSubscribed(newSubscription, username);