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

Moved processPubRel and processPubComp into ProtocolProcessor

Browse Source
This commit is contained in:
andrea
2013-08-30 13:23:06 +02:00
parent 82e998e939
commit 7d0cb70584
2 changed files with 34 additions and 37 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
+33
View File
@@ -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.events.PublishEvent;
import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; 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.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.AbstractMessage;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
@@ -260,5 +261,37 @@ class ProtocolProcessor {
LOG.error(null, t); 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);
}
} }
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+1 -37
View File
@@ -181,7 +181,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} else if (message instanceof PubRelMessage) { } else if (message instanceof PubRelMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubRelMessage) message).getMessageID(); int messageID = ((PubRelMessage) message).getMessageID();
processPubRel(clientID, messageID); m_processor.processPubRel(clientID, messageID);
} else if (message instanceof PubRecMessage) { } else if (message instanceof PubRecMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubRecMessage) message).getMessageID(); int messageID = ((PubRecMessage) message).getMessageID();
@@ -208,34 +208,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} }
// 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) { protected void processPublish(PublishEvent evt) {
m_processor.processPublish(evt); m_processor.processPublish(evt);
} }
@@ -342,12 +314,4 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
pubEvt.getMessage(), false, pubEvt.getMessageID())); 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);
}
} }