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

Added unsubscribe management for server side

Browse Source
This commit is contained in:
andrea
2012-08-19 12:17:30 +02:00
parent 54e0b1520d
commit f691767fd5
5 changed files with 105 additions and 14 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
@@ -29,4 +29,10 @@ public interface IMessaging {
void removeSubscriptions(String clientID);
void close();
/**
* Remove the clientID from topic subscription, if not previously subscribed,
* doesn't reply any error
*/
void unsubscribe(String topic, String clientID);
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+25 -12
View File
@@ -119,20 +119,33 @@ public class SimpleMessaging implements IMessaging {
public void subscribe(String clientId, String topic, QOSType qos) {
Subscription newSubscription = new Subscription(clientId, topic, qos);
rwLock.writeLock().lock();
subscriptions.add(newSubscription);
//scans reatained messages to be published to the new subscription
LOG.debug("Scanning all retained messages...");
for (Map.Entry<String, StoredMessage> entry : m_retainedStore) {
StoredMessage storedMsg = entry.getValue();
if (matchTopics(entry.getKey(), topic)) {
//fire the as retained the message
m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(),
storedMsg.getPayload(), true);
try {
subscriptions.add(newSubscription);
//scans retained messages to be published to the new subscription
LOG.debug("Scanning all retained messages...");
for (Map.Entry<String, StoredMessage> entry : m_retainedStore) {
StoredMessage storedMsg = entry.getValue();
if (matchTopics(entry.getKey(), topic)) {
//fire the as retained the message
m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(),
storedMsg.getPayload(), true);
}
}
} finally {
LOG.debug("Finished firing retained messages");
rwLock.writeLock().unlock();
}
}
public void unsubscribe(String topic, String clientID) {
rwLock.writeLock().lock();
try {
subscriptions.removeSubscription(topic, clientID);
} finally {
rwLock.writeLock().unlock();
}
LOG.debug("Finished firing retained messages");
rwLock.writeLock().unlock();
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java
+49 -2
View File
@@ -217,12 +217,42 @@ public class SubscriptionsStore {
}
protected void addDirect(Subscription newSubscription) {
// List<Token> tokens = new ArrayList<Token>();
// try {
// tokens = splitTopic(newSubscription.topic);
// } catch (ParseException ex) {
// //TODO handle the parse exception
// LOG.error(null, ex);
//// return;
// }
//
// TreeNode current = subscriptions;
// for (Token token : tokens) {
// TreeNode matchingChildren;
//
// //check if a children with the same token already exists
// if ((matchingChildren = current.childWithToken(token)) != null) {
// current = matchingChildren;
// } else {
// //create a new node for the newly inserted token
// matchingChildren = new TreeNode(current);
// matchingChildren.setToken(token);
// current.addChild(matchingChildren);
// current = matchingChildren;
// }
// }
TreeNode current = findMatchingNode(newSubscription.topic);
current.addSubcription(newSubscription);
}
private TreeNode findMatchingNode(String topic) {
List<Token> tokens = new ArrayList<Token>();
try {
tokens = splitTopic(newSubscription.topic);
tokens = splitTopic(topic);
} catch (ParseException ex) {
//TODO handle the parse exception
LOG.error(null, ex);
// return;
}
TreeNode current = subscriptions;
@@ -240,7 +270,7 @@ public class SubscriptionsStore {
current = matchingChildren;
}
}
current.addSubcription(newSubscription);
return current;
}
public void add(Subscription newSubscription) {
@@ -257,6 +287,23 @@ public class SubscriptionsStore {
m_persistent.put(clientID, subs);
// pageFile.flush();
}
public void removeSubscription(String topic, String clientID) {
TreeNode matchNode = findMatchingNode(topic);
//search fr the subscription to remove
Subscription toBeRemoved = null;
for (Subscription sub : matchNode.subscriptions()) {
if (sub.topic.equals(topic)) {
toBeRemoved = sub;
break;
}
}
if (toBeRemoved != null) {
matchNode.subscriptions().remove(toBeRemoved);
}
}
/**
* Visit the topics tree to remove matching subscriptions with clientID
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+20
View File
@@ -19,6 +19,8 @@ import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.SubscribeMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,6 +52,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
case SUBSCRIBE:
handleSubscribe(session, (SubscribeMessage) msg);
break;
case UNSUBSCRIBE:
handleUnsubscribe(session, (UnsubscribeMessage) msg);
break;
case PUBLISH:
handlePublish(session, (PublishMessage) msg);
break;
@@ -164,6 +169,19 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
LOG.info("replying with SubAct to MSG ID {0}", msg.getMessageID());
session.write(ackMessage);
}
private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) {
LOG.info("unregistering the subscriptions");
for (String topic : msg.topics()) {
m_messaging.unsubscribe(topic, (String) session.getAttribute(ATTR_CLIENTID));
}
//ack the client
UnsubAckMessage ackMessage = new UnsubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
LOG.info("replying with UnsubAct to MSG ID {0}", msg.getMessageID());
session.write(ackMessage);
}
protected void handlePublish(IoSession session, PublishMessage message) {
m_messaging.publish(message.getTopicName(), message.getPayload(),
@@ -205,6 +223,8 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier {
pubMessage.setTopicName(topic);
pubMessage.setQos(qOSType);
pubMessage.setPayload(payload);
assert m_clientIDs != null;
assert m_clientIDs.get(clientId) != null;
m_clientIDs.get(clientId).getSession().write(pubMessage);
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/Server.java
+5
View File
@@ -21,10 +21,13 @@ import org.dna.mqtt.moquette.proto.PublishDecoder;
import org.dna.mqtt.moquette.proto.PublishEncoder;
import org.dna.mqtt.moquette.proto.SubAckEncoder;
import org.dna.mqtt.moquette.proto.SubscribeDecoder;
import org.dna.mqtt.moquette.proto.UnsubAckEncoder;
import org.dna.mqtt.moquette.proto.UnsubscribeDecoder;
import org.dna.mqtt.moquette.proto.messages.ConnAckMessage;
import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
import org.dna.mqtt.moquette.proto.messages.PublishMessage;
import org.dna.mqtt.moquette.proto.messages.SubAckMessage;
import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
@@ -52,6 +55,7 @@ public class Server {
decoder.addMessageDecoder(new ConnectDecoder());
decoder.addMessageDecoder(new PublishDecoder());
decoder.addMessageDecoder(new SubscribeDecoder());
decoder.addMessageDecoder(new UnsubscribeDecoder());
decoder.addMessageDecoder(new DisconnectDecoder());
decoder.addMessageDecoder(new PingReqDecoder());
@@ -59,6 +63,7 @@ public class Server {
// encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder());
encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder());
encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder());
encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder());
encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder());