anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved processPubRec into ProtocolProcessor

Browse Source
This commit is contained in:
andrea
2013-08-30 13:24:53 +02:00
parent 7d0cb70584
commit e74f6d5655
2 changed files with 12 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+11 -1
View File
@@ -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);
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+1 -10
View File
@@ -185,7 +185,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} 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<ValueEvent> {
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);