Moved qos2 publish to separate class

2016-12-13 23:43:02 +01:00
parent 1c2fe8c8fe
commit 1e04292ba2
2 changed files with 134 additions and 98 deletions
@@ -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<String, ConnectionDescriptor> connectionDescriptors;
+    private final ISessionsStore m_sessionsStore;
+    private final IMessagesStore m_messagesStore;
+
+    public Qos2Publisher(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore,
+                         IMessagesStore messagesStore) {
+        this.connectionDescriptors = connectionDescriptors;
+        this.m_sessionsStore = sessionsStore;
+        m_messagesStore = messagesStore;
+    }
+
+    void publish2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> 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);
+        }
+    }
+
+}