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

Moved subscription logic inside the event loop

Browse Source
This commit is contained in:
selva.andre
2012-09-04 16:20:53 +02:00
parent 70ad0c0475
commit 0e1a3cef6c
5 changed files with 46 additions and 89 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
-5
View File
@@ -14,11 +14,6 @@ import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
*/
public interface IMessaging {
/**
* Subscribe a client to a specified topic with a defined level
*/
void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID);
void stop();
void disconnect(IoSession session);
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+27 -33
View File
@@ -14,7 +14,6 @@ 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;
@@ -115,13 +114,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID) {
Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession);
LOG.debug("subscribe invoked for topic: " + topic);
disruptorPublish(new SubscribeEvent(newSubscription, messageID));
}
public void disconnect(IoSession session) {
disruptorPublish(new DisconnectEvent(session));
}
@@ -178,9 +170,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
return subscriptions;
}
/* public void removeSubscriptions(String clientID) {
disruptorPublish(new RemoveAllSubscriptionsEvent(clientID));
}*/
public void stop() {
disruptorPublish(new StopEvent());
@@ -191,8 +180,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
MessagingEvent evt = t.getEvent();
if (evt instanceof PublishEvent) {
processPublish((PublishEvent) evt);
} else if (evt instanceof SubscribeEvent) {
processSubscribe((SubscribeEvent) evt);
} else if (evt instanceof StopEvent) {
processStop();
} else if (evt instanceof DisconnectEvent) {
@@ -237,7 +224,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
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);
processSubscribe(session, (SubscribeMessage) message, clientID, cleanSession);
} else {
throw new RuntimeException("Illegal message received " + message);
}
@@ -373,11 +363,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
}
protected void processSubscribe(SubscribeEvent evt) {
LOG.debug("processSubscribe invoked");
Subscription newSubscription = evt.getSubscription();
final String topic = newSubscription.getTopic();
private void subscribeSingleTopic(Subscription newSubscription, final String topic) {
subscriptions.add(newSubscription);
//scans retained messages to be published to the new subscription
@@ -390,22 +376,30 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
for (StoredMessage storedMsg : messages) {
//fire the as retained the message
LOG.debug("Inserting NotifyEvent into outbound for topic " + topic);
notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(),
storedMsg.getPayload(), true));
notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(), storedMsg.getPayload(), true));
}
}
protected void processSubscribe(IoSession session, SubscribeMessage msg, String clientID, boolean cleanSession) {
LOG.debug("processSubscribe invoked");
for (SubscribeMessage.Couple req : msg.subscriptions()) {
QOSType qos = AbstractMessage.QOSType.values()[req.getQos()];
Subscription newSubscription = new Subscription(clientID, req.getTopic(), qos, cleanSession);
subscribeSingleTopic(newSubscription, req.getTopic());
}
// //ack the client
// SubAckMessage ackMessage = new SubAckMessage();
// ackMessage.setMessageID(evt.getMessageID());
//
// //TODO by now it handles only QoS 0 messages
// /*for (int i = 0; i < evt.subscriptions().size(); i++) {
// ackMessage.addType(QOSType.MOST_ONE);
// } */
// LOG.info("replying with SubAct to MSG ID {0}", evt.getMessageID());
// //session.write(ackMessage);
// IoSession session = m_clientIDs.get(evt.getSubscription().clientId).getSession();
// session.write(ackMessage).awaitUninterruptibly();
//ack the client
SubAckMessage ackMessage = new SubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
//TODO by now it handles only QoS 0 messages
for (int i = 0; i < msg.subscriptions().size(); i++) {
ackMessage.addType(QOSType.MOST_ONE);
}
LOG.info("replying with SubAct to MSG ID {0}", msg.getMessageID());
session.write(ackMessage);
}
/**
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/ConnectEvent.java
-25
View File
@@ -1,25 +0,0 @@
package org.dna.mqtt.moquette.messaging.spi.impl.events;
import org.apache.mina.core.session.IoSession;
import org.dna.mqtt.moquette.proto.messages.ConnectMessage;
/**
*/
public class ConnectEvent extends MessagingEvent {
IoSession m_session;
ConnectMessage message;
public ConnectEvent(IoSession session, ConnectMessage message) {
this.m_session = session;
this.message = message;
}
public IoSession getSession() {
return m_session;
}
public ConnectMessage getMessage() {
return message;
}
}
parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java
+1 -17
View File
@@ -66,23 +66,7 @@ public class MQTTHandler extends IoHandlerAdapter {
protected void handleSubscribe(IoSession session, SubscribeMessage msg) {
LOG.debug("handleSubscribe, registering the subscriptions");
for (SubscribeMessage.Couple req : msg.subscriptions()) {
String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
m_messaging.subscribe(clientID, req.getTopic(), AbstractMessage.QOSType.values()[req.getQos()],
cleanSession, msg.getMessageID());
}
//ack the client
SubAckMessage ackMessage = new SubAckMessage();
ackMessage.setMessageID(msg.getMessageID());
//TODO by now it handles only QoS 0 messages
for (int i = 0; i < msg.subscriptions().size(); i++) {
ackMessage.addType(QOSType.MOST_ONE);
}
LOG.info("replying with SubAct to MSG ID {0}", msg.getMessageID());
session.write(ackMessage);
m_messaging.handleProtocolMessage(session, msg);
}
private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) {
parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessagingTest.java
+18 -9
View File
@@ -76,8 +76,9 @@ public class SimpleMessagingTest {
@Test
public void testSubscribe() {
//Exercise
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
SubscribeMessage msg = new SubscribeMessage();
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//Verify
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false);
@@ -86,11 +87,14 @@ public class SimpleMessagingTest {
@Test
public void testDoubleSubscribe() {
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
// SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
SubscribeMessage msg = new SubscribeMessage();
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//Exercise
messaging.processSubscribe(evt);
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//Verify
Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false);
@@ -103,8 +107,11 @@ public class SimpleMessagingTest {
ConnectionDescriptor connDescr = new ConnectionDescriptor(FAKE_CLIENT_ID, m_session, true);
messaging.m_clientIDs.put(FAKE_CLIENT_ID, connDescr);
SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
messaging.processSubscribe(evt);
// SubscribeEvent evt = new SubscribeEvent(new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, QOSType.MOST_ONE, false), 0);
SubscribeMessage msg = new SubscribeMessage();
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.MOST_ONE.ordinal(), FAKE_TOPIC));
messaging.processSubscribe(m_session, msg, FAKE_CLIENT_ID, false);
//Exercise
PublishEvent pubEvt = new PublishEvent(FAKE_TOPIC, QOSType.MOST_ONE, "Hello".getBytes(), false, "FakeCLI", null);
@@ -198,8 +205,10 @@ public class SimpleMessagingTest {
//Exercise
//m_handler.setMessaging(mockedMessaging);
SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0);
messaging.processSubscribe(pubEvt);
// SubscribeEvent pubEvt = new SubscribeEvent(subscription, 0);
SubscribeMessage msg = new SubscribeMessage();
msg.addSubscription(new SubscribeMessage.Couple((byte)QOSType.EXACTLY_ONCE.ordinal(), "/news"));
messaging.processSubscribe(m_session, msg, "fakeID", false);
//Verify
/*ArgumentCaptor<AbstractMessage.QOSType> argument = ArgumentCaptor.forClass(AbstractMessage.QOSType.class);