From ed6ad315d1de8eec0515284e2526bda929c991a9 Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 23 Mar 2016 23:13:01 +0100 Subject: [PATCH] Fixed merge --- .../moquette/spi/impl/ProtocolProcessor.java | 95 +++++++++++-------- 1 file changed, 54 insertions(+), 41 deletions(-) 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 287b66ba..99297786 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -57,9 +57,9 @@ import org.slf4j.LoggerFactory; /** * Class responsible to handle the logic of MQTT protocol it's the director of * the protocol execution. - * + * * Used by the front facing class SimpleMessaging. - * + * * @author andrea */ public class ProtocolProcessor { @@ -92,11 +92,11 @@ public class ProtocolProcessor { public QOSType getQos() { return qos; } - + } - + private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); - + protected ConcurrentMap m_clientIDs; private SubscriptionsStore subscriptions; private boolean allowAnonymous; @@ -108,7 +108,7 @@ public class ProtocolProcessor { //maps clientID to Will testament, if specified on CONNECT private ConcurrentMap m_willStore = new ConcurrentHashMap<>(); - + ProtocolProcessor() {} /** @@ -209,8 +209,9 @@ public class ProtocolProcessor { byte[] willPayload = msg.getWillMessage(); ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); //save the will testament in the clientID store - WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos ); + WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos); m_willStore.put(msg.getClientID(), will); + LOG.info("Session for clientID <{}> with will to topic {}", msg.getClientID(), msg.getWillTopic()); } ConnAckMessage okResp = new ConnAckMessage(); @@ -276,18 +277,18 @@ public class ProtocolProcessor { clientSession.removeEnqueued(pubEvt.getGuid()); } } - - public void processPubAck(Channel session, PubAckMessage msg) { - String clientID = NettyUtils.clientID(session); + + public void processPubAck(Channel channel, PubAckMessage msg) { + String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); - String username = NettyUtils.userName(session); - StoredMessage inflightMsg = m_sessionsStore.getInflightMessage(clientID, messageID); - + String username = NettyUtils.userName(channel); + StoredMessage inflightMsg = m_sessionsStore.getInflightMessage(clientID, messageID); + //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); targetSession.inFlightAcknowledged(messageID); - + String topic = inflightMsg.getTopic(); m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username)); @@ -311,13 +312,13 @@ public class ProtocolProcessor { pub.setRetained(will.isRetained()); return pub; } - - public void processPublish(Channel session, PublishMessage msg) { + + public void processPublish(Channel channel, PublishMessage msg) { LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg); - String clientID = NettyUtils.clientID(session); + String clientID = NettyUtils.clientID(channel); final String topic = msg.getTopicName(); //check if the topic can be wrote - String username = NettyUtils.userName(session); + String username = NettyUtils.userName(channel); if (!m_authorizator.canWrite(topic, username, clientID)) { LOG.debug("topic {} doesn't have write credentials", topic); return; @@ -397,7 +398,7 @@ public class ProtocolProcessor { } m_messagesStore.storeRetained(topic, guid); } - + /** * Specialized version to publish will testament message. */ @@ -466,15 +467,17 @@ public class ProtocolProcessor { } } - protected void directSend(ClientSession clientsession, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) { + 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); + 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)); @@ -484,31 +487,40 @@ public class ProtocolProcessor { pubMessage.setMessageID(messageID); } else { if (messageID != null) { - throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message with PacketIdentifier: " + messageID); + throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message " + + "with PacketIdentifier: " + messageID); } } if (m_clientIDs == null) { - throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be initialized, somewhere it's overwritten!!"); + throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be " + + "initialized, somewhere it's overwritten!!"); } - LOG.debug("clientIDs are {}", m_clientIDs); + //LOG.trace("clientIDs are {}", m_clientIDs); if (m_clientIDs.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, m_clientIDs)); + throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>", + clientId, m_clientIDs)); } Channel channel = m_clientIDs.get(clientId).channel; - LOG.debug("Session for clientId {} is {}", clientId, channel); - channel.writeAndFlush(pubMessage); + LOG.trace("Session for clientId {}", clientId); + if (channel.isWritable()) { + //if channel is writable don't enqueue + channel.write(pubMessage); + } else { + //enqueue to the client session + clientsession.enqueue(pubMessage); + } } - + 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); m_clientIDs.get(clientID).channel.writeAndFlush(pubRecMessage); } - + private void sendPubAck(String clientId, int messageID) { LOG.trace("sendPubAck invoked"); PubAckMessage pubAckMessage = new PubAckMessage(); @@ -527,7 +539,7 @@ public class ProtocolProcessor { LOG.error(null, t); } } - + /** * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish * to all interested subscribers. @@ -552,7 +564,7 @@ public class ProtocolProcessor { 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(); @@ -560,7 +572,7 @@ public class ProtocolProcessor { m_clientIDs.get(clientID).channel.writeAndFlush(pubCompMessage); } - + public void processPubRec(Channel channel, PubRecMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); @@ -581,8 +593,8 @@ public class ProtocolProcessor { public void processPubComp(Channel channel, PubCompMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); - StoredMessage inflightMsg = m_sessionsStore.getInflightMessage( clientID, messageID ); - + StoredMessage inflightMsg = m_sessionsStore.getInflightMessage( clientID, messageID ); + LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID); //once received the PUBCOMP then remove the message from the temp memory ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); @@ -592,8 +604,9 @@ public class ProtocolProcessor { String topic = inflightMsg.getTopic(); m_interceptor.notifyMessageAcknowledged( new InterceptAcknowledgedMessage(inflightMsg, topic, username) ); } - - public void processDisconnect(Channel channel) { + + public void processDisconnect(Channel channel) throws InterruptedException { + channel.flush(); String clientID = NettyUtils.clientID(channel); boolean cleanSession = NettyUtils.cleanSession(channel); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); @@ -605,7 +618,7 @@ public class ProtocolProcessor { //cleanup the will store m_willStore.remove(clientID); - + String username = NettyUtils.userName(channel); m_interceptor.notifyClientDisconnected(clientID, username); LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession); @@ -628,7 +641,7 @@ public class ProtocolProcessor { m_willStore.remove(clientID); } } - + /** * Remove the clientID from topic subscription, if not previously subscribed, * doesn't reply any error @@ -706,7 +719,7 @@ public class ProtocolProcessor { subscribeSingleTopic(subscription, username); } } - + private boolean subscribeSingleTopic(final Subscription newSubscription, String username) { subscriptions.add(newSubscription.asClientTopicCouple()); @@ -714,7 +727,7 @@ public class ProtocolProcessor { //TODO this is ugly, it does a linear scan on potential big dataset Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { @Override - public boolean match(String key) { + public boolean match(String key) { return SubscriptionsStore.matchTopics(key, newSubscription.getTopicFilter()); } });