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

Moved disconnect message handling inside event loop

Browse Source
This commit is contained in:
selva.andre
2012-09-04 15:14:16 +02:00
parent 342a6a6187
commit 598efd26f3
4 changed files with 24 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java
-6
View File
@@ -19,12 +19,6 @@ public interface IMessaging {
*/
void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID);
/**
* Remove all subscription to any topic the client (identified by clientID)
* was subscribed
*/
void removeSubscriptions(String clientID);
void stop();
/**
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+23 -14
View File
@@ -35,7 +35,7 @@ import org.slf4j.LoggerFactory;
*
* @author andrea
*/
public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*, Runnable*/ {
public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
//TODO probably move this
public static class StoredMessage implements Serializable {
@@ -190,9 +190,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
return subscriptions;
}
public void removeSubscriptions(String clientID) {
/* public void removeSubscriptions(String clientID) {
disruptorPublish(new RemoveAllSubscriptionsEvent(clientID));
}
}*/
public void stop() {
disruptorPublish(new StopEvent());
@@ -207,12 +207,12 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
processSubscribe((SubscribeEvent) evt);
} else if (evt instanceof UnsubscribeEvent) {
processUnsubscribe((UnsubscribeEvent) evt);
} else if (evt instanceof RemoveAllSubscriptionsEvent) {
processRemoveAllSubscriptions((RemoveAllSubscriptionsEvent) evt);
} else if (evt instanceof StopEvent) {
processStop();
} else if (evt instanceof DisconnectEvent) {
processDisconnect((DisconnectEvent) evt);
DisconnectEvent disEvt = (DisconnectEvent) evt;
String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
processDisconnect(disEvt.getSession(), clientID);
} else if (evt instanceof CleanInFlightEvent) {
//remove the message from inflight storage
m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId());
@@ -236,6 +236,17 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
pubEvt = new PublishEvent(pubMsg.getTopicName(), pubMsg.getQos(), pubMsg.getPayload(), pubMsg.isRetainFlag(), clientID, pubMsg.getMessageID(), session);
}
processPublish(pubEvt);
} else if (message instanceof DisconnectMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
processRemoveAllSubscriptions(clientID);
}
//close the TCP connection
//session.close(true);
processDisconnect(session, clientID);
} else {
throw new RuntimeException("Illegal message received " + message);
}
@@ -276,7 +287,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
removeSubscriptions(msg.getClientID());
processRemoveAllSubscriptions(msg.getClientID());
}
m_clientIDs.get(msg.getClientID()).getSession().close(false);
@@ -319,7 +330,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
if (msg.isCleanSession()) {
//remove all prev subscriptions
//cleanup topic subscriptions
removeSubscriptions(msg.getClientID());
processRemoveAllSubscriptions(msg.getClientID());
} else {
//force the republish of stored QoS1 and QoS2
republishStored(msg.getClientID());
@@ -411,18 +422,16 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent>/*,
subscriptions.removeSubscription(evt.getTopic(), evt.getClientID());
}
protected void processRemoveAllSubscriptions(RemoveAllSubscriptionsEvent evt) {
protected void processRemoveAllSubscriptions(String clientID) {
LOG.debug("processRemoveAllSubscriptions invoked");
subscriptions.removeForClient(evt.getClientID());
subscriptions.removeForClient(clientID);
//remove also the messages stored of type QoS1/2
m_storageService.cleanPersistedPublishes(evt.getClientID());
m_storageService.cleanPersistedPublishes(clientID);
}
private void processDisconnect(DisconnectEvent evt) throws InterruptedException {
private void processDisconnect(IoSession session, String clientID) throws InterruptedException {
// m_notifier.disconnect(evt.getSession());
IoSession session = evt.getSession();
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
m_clientIDs.remove(clientID);
session.close(true);
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/RemoveAllSubscriptionsEvent.java
-19
View File
@@ -1,19 +0,0 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
/**
*
* @author andrea
*/
public class RemoveAllSubscriptionsEvent extends MessagingEvent {
String m_clientID;
public RemoveAllSubscriptionsEvent(String clientID) {
m_clientID = clientID;
}
public String getClientID() {
return m_clientID;
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+1 -10
View File
@@ -103,16 +103,7 @@ public class MQTTHandler extends IoHandlerAdapter {
}
protected void handleDisconnect(IoSession session, DisconnectMessage disconnectMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
if (cleanSession) {
//cleanup topic subscriptions
m_messaging.removeSubscriptions(clientID);
}
//close the TCP connection
//session.close(true);
m_messaging.disconnect(session);
m_messaging.handleProtocolMessage(session, disconnectMessage);
}
@Override