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

Moved Unsubscribe message handling inside event loop

Browse Source
This commit is contained in:
selva.andre
2012-09-04 15:42:37 +02:00
parent 598efd26f3
commit d6f6892d37
3 changed files with 27 additions and 9 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
+1 -1
View File
@@ -25,7 +25,7 @@ public interface IMessaging {
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
*/
void unsubscribe(String topic, String clientID);
//void unsubscribe(String topic, String clientID);
void disconnect(IoSession session);
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+21 -6
View File
@@ -14,6 +14,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.util.SessionAttributeInitializingFilter;
import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.IStorageService;
@@ -129,9 +130,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
public void unsubscribe(String topic, String clientID) {
/*public void unsubscribe(String topic, String clientID) {
disruptorPublish(new UnsubscribeEvent(topic, clientID));
}
} */
public void disconnect(IoSession session) {
@@ -205,9 +206,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
processPublish((PublishEvent) evt);
} else if (evt instanceof SubscribeEvent) {
processSubscribe((SubscribeEvent) evt);
} else if (evt instanceof UnsubscribeEvent) {
} /*else if (evt instanceof UnsubscribeEvent) {
processUnsubscribe((UnsubscribeEvent) evt);
} else if (evt instanceof StopEvent) {
} */else if (evt instanceof StopEvent) {
processStop();
} else if (evt instanceof DisconnectEvent) {
DisconnectEvent disEvt = (DisconnectEvent) evt;
@@ -247,6 +248,11 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
//close the TCP connection
//session.close(true);
processDisconnect(session, clientID);
} else if (message instanceof UnsubscribeMessage) {
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
} else {
throw new RuntimeException("Illegal message received " + message);
}
@@ -417,9 +423,18 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
// session.write(ackMessage).awaitUninterruptibly();
}
protected void processUnsubscribe(UnsubscribeEvent evt) {
protected void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
LOG.debug("processSubscribe invoked");
subscriptions.removeSubscription(evt.getTopic(), evt.getClientID());
for (String topic : topics) {
subscriptions.removeSubscription(topic, clientID);
}
//ack the client
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(messageID);
LOG.info("replying with UnsubAck to MSG ID {0}", messageID);
session.write(ackMessage);
}
protected void processRemoveAllSubscriptions(String clientID) {
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+5 -2
View File
@@ -87,7 +87,10 @@ public class MQTTHandler extends IoHandlerAdapter {
private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) {
LOG.info("unregistering the subscriptions");
for (String topic : msg.topics()) {
m_messaging.handleProtocolMessage(session, msg);
/*for (String topic : msg.topics()) {
m_messaging.unsubscribe(topic, (String) session.getAttribute(Constants.ATTR_CLIENTID));
}
//ack the client
@@ -95,7 +98,7 @@ public class MQTTHandler extends IoHandlerAdapter {
ackMessage.setMessageID(msg.getMessageID());
LOG.info("replying with UnsubAck to MSG ID {0}", msg.getMessageID());
session.write(ackMessage);
session.write(ackMessage); */
}
protected void handlePublish(IoSession session, PublishMessage message) {