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 84bb899d..8e02addc 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 @@ -10,6 +10,7 @@ import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; +import org.dna.mqtt.moquette.proto.PubCompMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; @@ -260,5 +261,37 @@ 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. + * */ + protected void processPubRel(String clientID, int messageID) { + String publishKey = String.format("%s%d", clientID, messageID); + PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); + + final String topic = evt.getTopic(); + final AbstractMessage.QOSType qos = evt.getQos(); + final byte[] message = evt.getMessage(); + boolean retain = evt.isRetain(); + + publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + + m_storageService.removeQoS2Message(publishKey); + + if (retain) { + m_storageService.storeRetained(topic, message, qos); + } + + sendPubComp(clientID, messageID); + } + + private void sendPubComp(String clientID, int messageID) { + LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); + PubCompMessage pubCompMessage = new PubCompMessage(); + pubCompMessage.setMessageID(messageID); + + m_clientIDs.get(clientID).getSession().write(pubCompMessage); + } } 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 031f987f..90ce7bce 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 @@ -181,7 +181,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } else if (message instanceof PubRelMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = ((PubRelMessage) message).getMessageID(); - processPubRel(clientID, messageID); + m_processor.processPubRel(clientID, messageID); } else if (message instanceof PubRecMessage) { String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); int messageID = ((PubRecMessage) message).getMessageID(); @@ -208,34 +208,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { } -// protected void processConnect(IoSession session, ConnectMessage msg) { -// m_processor.processConnect(session, msg); -// } - - /** - * 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) { - String publishKey = String.format("%s%d", clientID, messageID); - PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); - - final String topic = evt.getTopic(); - final QOSType qos = evt.getQos(); - final byte[] message = evt.getMessage(); - boolean retain = evt.isRetain(); - - m_processor.publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); - - m_storageService.removeQoS2Message(publishKey); - - if (retain) { - m_storageService.storeRetained(topic, message, qos); - } - - sendPubComp(clientID, messageID); - } - protected void processPublish(PublishEvent evt) { m_processor.processPublish(evt); } @@ -342,12 +314,4 @@ public class SimpleMessaging implements IMessaging, EventHandler { pubEvt.getMessage(), false, pubEvt.getMessageID())); } } - - private void sendPubComp(String clientID, int messageID) { - LOG.debug(String.format("sendPubComp invoked for clientID %s ad messageID %d", clientID, messageID)); - PubCompMessage pubCompMessage = new PubCompMessage(); - pubCompMessage.setMessageID(messageID); - - m_clientIDs.get(clientID).getSession().write(pubCompMessage); - } }