From 12ed805737267de55cc732c91b5b7c6af1189d5e Mon Sep 17 00:00:00 2001 From: andsel Date: Fri, 9 Dec 2016 15:09:50 +0100 Subject: [PATCH] Moved qos0 publishing logic to separate class --- .../moquette/spi/impl/ProtocolProcessor.java | 76 ++++------ .../moquette/spi/impl/Qos0PublishHandler.java | 137 ++++++++++++++++++ 2 files changed, 165 insertions(+), 48 deletions(-) create mode 100644 broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.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 b02522a3..7f9e2382 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -133,6 +133,8 @@ public class ProtocolProcessor { private BrokerInterceptor m_interceptor; private String m_server_port; + private Qos0PublishHandler qos0PublishHandler; + //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); @@ -182,6 +184,9 @@ public class ProtocolProcessor { m_messagesStore = storageService; m_sessionsStore = sessionsStore; m_server_port = serverPort; + + this.qos0PublishHandler = new Qos0PublishHandler(m_authorizator, subscriptions, m_messagesStore, + m_interceptor, this.connectionDescriptors, m_sessionsStore); } public void processConnect(Channel channel, ConnectMessage msg) { @@ -421,7 +426,7 @@ public class ProtocolProcessor { m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username)); } - private static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) { + static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) { IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName()); stored.setRetained(msg.isRetainFlag()); stored.setMessageID(msg.getMessageID()); @@ -439,7 +444,7 @@ public class ProtocolProcessor { final AbstractMessage.QOSType qos = msg.getQos(); switch (qos) { case MOST_ONE: - receivedPublishQos0(channel, msg); + this.qos0PublishHandler.receivedPublishQos0(channel, msg); break; case LEAST_ONE: receivedPublishQos1(channel, msg); @@ -450,34 +455,9 @@ public class ProtocolProcessor { } } - private void receivedPublishQos0(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); - - if (msg.isRetainFlag()) { - //QoS == 0 && retain => clean old retained - m_messagesStore.cleanRetained(topic); - } - - String username = NettyUtils.userName(channel); - m_interceptor.notifyTopicPublished(msg, clientID, username); - } - private void receivedPublishQos1(Channel channel, PublishMessage msg) { //verify if topic can be write final String topic = msg.getTopicName(); - //check if the topic can be wrote if (checkWriteOnTopic(topic, channel)) { return; } @@ -542,7 +522,7 @@ public class ProtocolProcessor { m_interceptor.notifyTopicPublished(msg, clientID, username); } - private boolean checkWriteOnTopic(String topic, Channel channel) { + boolean checkWriteOnTopic(String topic, Channel channel) { String clientID = NettyUtils.clientID(channel); String username = NettyUtils.userName(channel); if (!m_authorizator.canWrite(topic, username, clientID)) { @@ -634,10 +614,7 @@ public class ProtocolProcessor { LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic); for (final Subscription sub : topicMatchingSubscriptions) { - AbstractMessage.QOSType qos = publishingQos; - if (qos.byteValue() > sub.getRequestedQos().byteValue()) { - qos = sub.getRequestedQos(); - } + AbstractMessage.QOSType qos = lowerQosToTheSubscriptionDesired(sub, publishingQos); ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); boolean targetIsActive = this.connectionDescriptors.containsKey(sub.getClientId()); @@ -645,27 +622,30 @@ public class ProtocolProcessor { 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); - } + //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(); + } + return qos; + } + protected void directSend(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) { String clientId = clientsession.clientID; diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java new file mode 100644 index 00000000..52c2d709 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/impl/Qos0PublishHandler.java @@ -0,0 +1,137 @@ +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.server.netty.NettyUtils; +import io.moquette.spi.ClientSession; +import io.moquette.spi.IMessagesStore; +import io.moquette.spi.ISessionsStore; +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; + +public class Qos0PublishHandler { + + private static final Logger LOG = LoggerFactory.getLogger(Qos0PublishHandler.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; + + public Qos0PublishHandler(IAuthorizator authorizator, SubscriptionsStore subscriptions, + IMessagesStore messagesStore, BrokerInterceptor interceptor, ConcurrentMap connectionDescriptors, ISessionsStore sessionsStore) { + this.m_authorizator = authorizator; + this.subscriptions = subscriptions; + this.m_messagesStore = messagesStore; + this.m_interceptor = interceptor; + this.connectionDescriptors = connectionDescriptors; + this.m_sessionsStore = sessionsStore; + } + + void receivedPublishQos0(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_qos0(toStoreMsg, topicMatchingSubscriptions); + + if (msg.isRetainFlag()) { + //QoS == 0 && retain => clean old retained + m_messagesStore.cleanRetained(topic); + } + + 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_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); + if (LOG.isTraceEnabled()) { + LOG.trace("content <{}>", DebugUtils.payload2Str(origMessage)); + LOG.trace("subscription tree {}", subscriptions.dumpTree()); + } + + LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic); + for (final Subscription sub : topicMatchingSubscriptions) { + boolean targetIsActive = connectionDescriptors.containsKey(sub.getClientId()); + + LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", + sub.getClientId(), sub.getTopicFilter(), AbstractMessage.QOSType.MOST_ONE, targetIsActive); + if (targetIsActive) { + ClientSession targetSession = m_sessionsStore.sessionForClient(sub.getClientId()); + ByteBuffer message = origMessage.duplicate(); + publishQos0(targetSession, topic, message); + } + } + } + + protected void publishQos0(ClientSession clientsession, String topic, ByteBuffer message) { + String clientId = clientsession.clientID; + + LOG.debug("publishQos0 invoked clientId <{}> on topic <{}>", clientId, topic); + PublishMessage pubMessage = new PublishMessage(); + pubMessage.setRetainFlag(false); + pubMessage.setTopicName(topic); + pubMessage.setQos(AbstractMessage.QOSType.MOST_ONE); + pubMessage.setPayload(message); + + 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; + //TODO attention channel could be null, because in the mean time it get closed + + LOG.trace("Session for clientId {}", clientId); + if (channel.isWritable()) { + LOG.debug("channel is writable"); + //if channel is writable don't enqueue + channel.writeAndFlush(pubMessage); + } + } +}