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

Moved processPubComp into ProtocolProcessor

Browse Source
This commit is contained in:
andrea
2013-08-30 13:28:55 +02:00
parent e74f6d5655
commit 353b70d6c7
2 changed files with 7 additions and 12 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
+6
View File
@@ -303,5 +303,11 @@ class ProtocolProcessor {
m_clientIDs.get(clientID).getSession().write(pubRelMessage);
}
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);
m_storageService.cleanInFlight(publishKey);
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+1 -12
View File
@@ -81,11 +81,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
/* public void setNotifier(INotifier notifier) {
m_notifier= notifier;
} */
private void disruptorPublish(MessagingEvent msgEvent) {
long sequence = m_ringBuffer.next();
ValueEvent event = m_ringBuffer.get(sequence);
@@ -189,7 +184,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} else if (message instanceof PubCompMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
int messageID = ((PubCompMessage) message).getMessageID();
processPubComp(clientID, messageID);
m_processor.processPubComp(clientID, messageID);
} else {
throw new RuntimeException("Illegal message received " + message);
}
@@ -212,12 +207,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
m_processor.processPublish(evt);
}
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);
m_storageService.cleanInFlight(publishKey);
}
private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
subscriptions.add(newSubscription);