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

Moved processDisconnect into ProtocolProcessor

Browse Source
This commit is contained in:
andrea
2013-08-30 14:53:57 +02:00
parent 353b70d6c7
commit b3293a031f
2 changed files with 12 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
+10
View File
@@ -310,4 +310,14 @@ class ProtocolProcessor {
m_storageService.cleanInFlight(publishKey); m_storageService.cleanInFlight(publishKey);
} }
void processDisconnect(IoSession session, String clientID) throws InterruptedException {
// m_notifier.disconnect(evt.getSession());
m_clientIDs.remove(clientID);
session.close(true);
//de-activate the subscriptions for this ClientID
// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
subscriptions.deactivate(clientID);
}
} }
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+2 -12
View File
@@ -133,7 +133,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} else if (evt instanceof DisconnectEvent) { } else if (evt instanceof DisconnectEvent) {
DisconnectEvent disEvt = (DisconnectEvent) evt; DisconnectEvent disEvt = (DisconnectEvent) evt;
String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID); String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
processDisconnect(disEvt.getSession(), clientID); m_processor.processDisconnect(disEvt.getSession(), clientID);
} else if (evt instanceof RepublishEvent) { } else if (evt instanceof RepublishEvent) {
processRepublish((RepublishEvent) evt); processRepublish((RepublishEvent) evt);
} else if (evt instanceof ProtocolEvent) { } else if (evt instanceof ProtocolEvent) {
@@ -164,7 +164,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
//close the TCP connection //close the TCP connection
//session.close(true); //session.close(true);
processDisconnect(session, clientID); m_processor.processDisconnect(session, clientID);
} else if (message instanceof UnsubscribeMessage) { } else if (message instanceof UnsubscribeMessage) {
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
@@ -263,16 +263,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
session.write(ackMessage); session.write(ackMessage);
} }
private void processDisconnect(IoSession session, String clientID) throws InterruptedException {
// m_notifier.disconnect(evt.getSession());
m_clientIDs.remove(clientID);
session.close(true);
//de-activate the subscriptions for this ClientID
// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
subscriptions.deactivate(clientID);
}
private void processStop() { private void processStop() {
LOG.debug("processStop invoked"); LOG.debug("processStop invoked");
m_storageService.close(); m_storageService.close();