From e74f6d565554b4a68172ac41aba44f9fe24ec925 Mon Sep 17 00:00:00 2001 From: andrea Date: Fri, 30 Aug 2013 13:24:53 +0200 Subject: [PATCH] Moved processPubRec into ProtocolProcessor --- .../messaging/spi/impl/ProtocolProcessor.java | 12 +++++++++++- .../moquette/messaging/spi/impl/SimpleMessaging.java | 11 +---------- 2 files changed, 12 insertions(+), 11 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 8e02addc..56a65079 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -16,6 +16,7 @@ import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.PubAckMessage; import org.dna.mqtt.moquette.proto.messages.PubRecMessage; +import org.dna.mqtt.moquette.proto.messages.PubRelMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.server.ConnectionDescriptor; import org.dna.mqtt.moquette.server.Constants; @@ -266,7 +267,7 @@ class ProtocolProcessor { * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish * to all interested subscribers. * */ - protected void processPubRel(String clientID, int messageID) { + void processPubRel(String clientID, int messageID) { String publishKey = String.format("%s%d", clientID, messageID); PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); @@ -293,5 +294,14 @@ class ProtocolProcessor { m_clientIDs.get(clientID).getSession().write(pubCompMessage); } + + void processPubRec(String clientID, int messageID) { + //once received a PUBREC reply with a PUBREL(messageID) + LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); + PubRelMessage pubRelMessage = new PubRelMessage(); + pubRelMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubRelMessage); + } } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 90ce7bce..a5c11aa9 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -185,7 +185,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else if (message instanceof PubRecMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = ((PubRecMessage) message).getMessageID(); - processPubRec(clientID, messageID); + m_processor.processPubRec(clientID, messageID); } else if (message instanceof PubCompMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = ((PubCompMessage) message).getMessageID(); @@ -212,15 +212,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_processor.processPublish(evt); } - private void processPubRec(String clientID, int messageID) { - //once received a PUBREC reply with a PUBREL(messageID) - LOG.debug(String.format("processPubRec invoked for clientID %s ad messageID %d", clientID, messageID)); - PubRelMessage pubRelMessage = new PubRelMessage(); - pubRelMessage.setMessageID(messageID); - - m_clientIDs.get(clientID).getSession().write(pubRelMessage); - } - private void processPubComp(String clientID, int messageID) { //once received the PUBCOMP then remove the message from the temp memory String publishKey = String.format("%s%d", clientID, messageID);