From d65886343b19ebd96571f46235c798f41dac8aec Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 11 Dec 2016 10:33:49 +0100 Subject: [PATCH] Moved QoS1 publish handling to separate class --- .../moquette/spi/impl/ProtocolProcessor.java | 86 +++----- .../moquette/spi/impl/Qos0PublishHandler.java | 2 +- .../moquette/spi/impl/Qos1PublishHandler.java | 191 ++++++++++++++++++ 3 files changed, 219 insertions(+), 60 deletions(-) create mode 100644 broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.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 e7c5f891..39611252 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -91,11 +91,11 @@ public class ProtocolProcessor { private class RunningSubscription { final String clientID; - final long packeId; + final long packetId; RunningSubscription(String clientID, long packeId) { this.clientID = clientID; - this.packeId = packeId; + this.packetId = packeId; } @Override @@ -105,7 +105,7 @@ public class ProtocolProcessor { RunningSubscription that = (RunningSubscription) o; - if (packeId != that.packeId) return false; + if (packetId != that.packetId) return false; return clientID != null ? clientID.equals(that.clientID) : that.clientID == null; } @@ -113,7 +113,7 @@ public class ProtocolProcessor { @Override public int hashCode() { int result = clientID != null ? clientID.hashCode() : 0; - result = 31 * result + (int) (packeId ^ (packeId >>> 32)); + result = 31 * result + (int) (packetId ^ (packetId >>> 32)); return result; } } @@ -134,6 +134,7 @@ public class ProtocolProcessor { private String m_server_port; private Qos0PublishHandler qos0PublishHandler; + private Qos1PublishHandler qos1PublishHandler; //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); @@ -187,6 +188,8 @@ public class ProtocolProcessor { this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, subscriptions, m_messagesStore, m_interceptor, this.connectionDescriptors, m_sessionsStore); + this.qos1PublishHandler = new Qos1PublishHandler(m_authorizator, subscriptions, m_messagesStore, + m_interceptor, this.connectionDescriptors, m_sessionsStore, m_server_port); } public void processConnect(Channel channel, ConnectMessage msg) { @@ -447,7 +450,7 @@ public class ProtocolProcessor { this.qos0PublishHandler.receivedPublishQos0(channel, msg); break; case LEAST_ONE: - receivedPublishQos1(channel, msg); + this.qos1PublishHandler.receivedPublishQos1(channel, msg); break; case EXACTLY_ONCE: receivedPublishQos2(channel, msg); @@ -455,41 +458,6 @@ public class ProtocolProcessor { } } - private void receivedPublishQos1(Channel channel, PublishMessage msg) { - //verify if topic can be write - final String topic = msg.getTopicName(); - if (checkWriteOnTopic(topic, channel)) { - return; - } - - //route message to subscribers - IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); - String clientID = NettyUtils.clientID(channel); - toStoreMsg.setClientID(clientID); - List topicMatchingSubscriptions = subscriptions.matches(topic); - publish2Subscribers(toStoreMsg, topicMatchingSubscriptions); - - //send PUBACK - final Integer messageID = msg.getMessageID(); - if (msg.isLocal()) { - sendPubAck(clientID, messageID); - } - LOG.info("server {} replying with PubAck to MSG ID {}", m_server_port, messageID); - - if (msg.isRetainFlag()) { - if (!msg.getPayload().hasRemaining()) { - m_messagesStore.cleanRetained(topic); - } else { - //before wasn't stored - MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); - m_messagesStore.storeRetained(topic, guid); - } - } - - String username = NettyUtils.userName(channel); - m_interceptor.notifyTopicPublished(msg, clientID, username); - } - private void receivedPublishQos2(Channel channel, PublishMessage msg) { final AbstractMessage.QOSType qos = QOSType.EXACTLY_ONCE; final String topic = msg.getTopicName(); @@ -608,7 +576,7 @@ public class ProtocolProcessor { } //if QoS 1 or 2 store the message MessageGUID guid = null; - if (publishingQos == QOSType.EXACTLY_ONCE || publishingQos == QOSType.LEAST_ONE) { + if (publishingQos != AbstractMessage.QOSType.MOST_ONE) { guid = m_messagesStore.storePublishForFuture(pubMsg); } @@ -706,24 +674,24 @@ public class ProtocolProcessor { connectionDescriptors.get(clientID).channel.writeAndFlush(pubRecMessage); } - private void sendPubAck(String clientId, int messageID) { - LOG.trace("sendPubAck invoked"); - PubAckMessage pubAckMessage = new PubAckMessage(); - pubAckMessage.setMessageID(messageID); - - try { - if (connectionDescriptors == null) { - throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be initialized, somewhere it's overwritten!!"); - } - LOG.debug("clientIDs are {}", connectionDescriptors); - if (connectionDescriptors.get(clientId) == null) { - throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, connectionDescriptors)); - } - connectionDescriptors.get(clientId).channel.writeAndFlush(pubAckMessage); - } catch(Throwable t) { - LOG.error(null, t); - } - } +// private static void sendPubAck(String clientId, int messageID) { +// LOG.trace("sendPubAck invoked"); +// PubAckMessage pubAckMessage = new PubAckMessage(); +// pubAckMessage.setMessageID(messageID); +// +// try { +// if (connectionDescriptors == null) { +// throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be initialized, somewhere it's overwritten!!"); +// } +// LOG.debug("clientIDs are {}", connectionDescriptors); +// if (connectionDescriptors.get(clientId) == null) { +// throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, connectionDescriptors)); +// } +// connectionDescriptors.get(clientId).channel.writeAndFlush(pubAckMessage); +// } catch(Throwable t) { +// LOG.error(null, t); +// } +// } /** * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java index 52c2d709..5bd02984 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -76,7 +76,7 @@ public class Qos0PublishHandler { return false; } - void publish2Subscribers_qos0(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { + private void publish2Subscribers_qos0(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { final String topic = pubMsg.getTopic(); final ByteBuffer origMessage = pubMsg.getMessage(); LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic {}", topic); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java new file mode 100644 index 00000000..2b204216 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -0,0 +1,191 @@ +package io.moquette.spi.impl; + +import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.parser.proto.messages.PubAckMessage; +import io.moquette.parser.proto.messages.PublishMessage; +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 Qos1PublishHandler { + 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 Qos1PublishHandler(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 receivedPublishQos1(Channel channel, PublishMessage msg) { + //verify if topic can be write + final String topic = msg.getTopicName(); + if (checkWriteOnTopic(topic, channel)) { + return; + } + + //route message to subscribers + IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); + String clientID = NettyUtils.clientID(channel); + toStoreMsg.setClientID(clientID); + List topicMatchingSubscriptions = subscriptions.matches(topic); + publish2Subscribers_qos1(toStoreMsg, topicMatchingSubscriptions); + + //send PUBACK + final Integer messageID = msg.getMessageID(); + if (msg.isLocal()) { + sendPubAck(clientID, messageID); + } + LOG.info("server {} replying with PubAck to MSG ID {}", brokerPort, messageID); + + if (msg.isRetainFlag()) { + if (!msg.getPayload().hasRemaining()) { + m_messagesStore.cleanRetained(topic); + } else { + //before wasn't stored + MessageGUID guid = m_messagesStore.storePublishForFuture(toStoreMsg); + 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; + } + + void publish2Subscribers_qos1(IMessagesStore.StoredMessage pubMsg, List topicMatchingSubscriptions) { + final String topic = pubMsg.getTopic(); + final AbstractMessage.QOSType publishingQos = pubMsg.getQos(); + final ByteBuffer origMessage = pubMsg.getMessage(); + LOG.debug("publish2Subscribers_qos1 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 = 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(); + //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); + publishQos1(targetSession, topic, message, messageId, qos); + } + } + } + } + + private void sendPubAck(String clientId, int messageID) { + LOG.trace("sendPubAck invoked"); + PubAckMessage pubAckMessage = new PubAckMessage(); + pubAckMessage.setMessageID(messageID); + + try { + if (connectionDescriptors == null) { + throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be initialized, somewhere it's overwritten!!"); + } + LOG.debug("clientIDs are {}", connectionDescriptors); + if (connectionDescriptors.get(clientId) == null) { + throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, connectionDescriptors)); + } + connectionDescriptors.get(clientId).channel.writeAndFlush(pubAckMessage); + } catch(Throwable t) { + LOG.error(null, t); + } + } + + protected void publishQos1(ClientSession clientsession, String topic, + ByteBuffer message, int messageID, AbstractMessage.QOSType qos) { + String clientId = clientsession.clientID; + LOG.debug("publishQos1 invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", + clientId, topic, qos, false, messageID); + PublishMessage pubMessage = new PublishMessage(); + pubMessage.setRetainFlag(false); + pubMessage.setTopicName(topic); + pubMessage.setQos(qos); + pubMessage.setPayload(message); + //QoS > 0 => set the PacketIdentifier + pubMessage.setMessageID(messageID); + + LOG.info("send publish message to <{}> on topic <{}>", clientId, topic); + if (LOG.isDebugEnabled()) { + LOG.debug("content <{}>", DebugUtils.payload2Str(message)); + } + + 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); + } + } +}