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

Moved processUnsubscribe into ProtocolProcessor

Browse Source
This commit is contained in:
andrea
2013-08-30 15:00:51 +02:00
parent b3293a031f
commit 90e2277490
2 changed files with 38 additions and 18 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
+20
View File
@@ -1,5 +1,6 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.util.List;
import java.util.Map;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
@@ -18,6 +19,7 @@ import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.dna.mqtt.moquette.server.ConnectionDescriptor;
import org.dna.mqtt.moquette.server.Constants;
import org.dna.mqtt.moquette.server.IAuthenticator;
@@ -319,5 +321,23 @@ class ProtocolProcessor {
// String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
subscriptions.deactivate(clientID);
}
/**
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
*/
void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
LOG.debug("processSubscribe invoked");
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);
}
}
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+18 -18
View File
@@ -168,7 +168,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
} else if (message instanceof UnsubscribeMessage) {
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
m_processor.processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
} else if (message instanceof SubscribeMessage) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
@@ -245,23 +245,23 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
session.write(ackMessage);
}
/**
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
*/
protected void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
LOG.debug("processSubscribe invoked");
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);
}
// /**
// * Remove the clientID from topic subscription, if not previously subscribed,
// * doesn't reply any error
// */
// protected void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
// LOG.debug("processSubscribe invoked");
//
// 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);
// }
private void processStop() {
LOG.debug("processStop invoked");