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

Moved selection of subscriptions out of route2Subscribers

Browse Source
This commit is contained in:
andsel
2016-12-13 23:43:01 +01:00
parent c727ee87d3
commit 3355e7c334
2 changed files with 19 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+14 -8
View File
@@ -461,7 +461,9 @@ public class ProtocolProcessor {
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
route2Subscribers(toStoreMsg);
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
route2Subscribers(toStoreMsg, topicMatchingSubscriptions);
if (msg.isRetainFlag()) {
//QoS == 0 && retain => clean old retained
@@ -494,7 +496,8 @@ public class ProtocolProcessor {
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
String clientID = NettyUtils.clientID(channel);
toStoreMsg.setClientID(clientID);
route2Subscribers(toStoreMsg);
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
route2Subscribers(toStoreMsg, topicMatchingSubscriptions);
//send PUBACK
final Integer messageID = msg.getMessageID();
@@ -573,7 +576,8 @@ public class ProtocolProcessor {
if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
guid = m_messagesStore.storePublishForFuture(toStoreMsg);
}
route2Subscribers(toStoreMsg);
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
route2Subscribers(toStoreMsg, topicMatchingSubscriptions);
if (!msg.isRetainFlag()) {
return;
@@ -604,14 +608,16 @@ public class ProtocolProcessor {
IMessagesStore.StoredMessage tobeStored = asStoredMessage(will);
tobeStored.setClientID(clientID);
tobeStored.setMessageID(messageId);
route2Subscribers(tobeStored);
String topic = tobeStored.getTopic();
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
route2Subscribers(tobeStored, topicMatchingSubscriptions);
}
/**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
void route2Subscribers(IMessagesStore.StoredMessage pubMsg) {
void route2Subscribers(IMessagesStore.StoredMessage pubMsg, List<Subscription> topicMatchingSubscriptions) {
final String topic = pubMsg.getTopic();
final AbstractMessage.QOSType publishingQos = pubMsg.getQos();
final ByteBuffer origMessage = pubMsg.getMessage();
@@ -626,7 +632,6 @@ public class ProtocolProcessor {
guid = m_messagesStore.storePublishForFuture(pubMsg);
}
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic);
for (final Subscription sub : topicMatchingSubscriptions) {
AbstractMessage.QOSType qos = publishingQos;
@@ -745,10 +750,11 @@ public class ProtocolProcessor {
LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
route2Subscribers(evt);
final String topic = evt.getTopic();
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
route2Subscribers(evt, topicMatchingSubscriptions);
if (evt.isRetained()) {
final String topic = evt.getTopic();
if (!evt.getMessage().hasRemaining()) {
m_messagesStore.cleanRetained(topic);
} else {
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+5 -3
View File
@@ -452,7 +452,8 @@ public class ProtocolProcessorTest {
@Test
public void testForwardPublishWithCorrectQos() {
StoredMessage forwardPublish = new StoredMessage("Hello world MQTT!!".getBytes(), QOSType.EXACTLY_ONCE, "a/b");
String topic = "a/b";
StoredMessage forwardPublish = new StoredMessage("Hello world MQTT!!".getBytes(), QOSType.EXACTLY_ONCE, topic);
forwardPublish.setRetained(true);
forwardPublish.setMessageID(1);
@@ -463,7 +464,7 @@ public class ProtocolProcessorTest {
sessionsStore.createNewSession("Sub A", false);
sessionsStore.createNewSession("Sub B", false);
Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE);
Subscription subQos1 = new Subscription("Sub A", topic, QOSType.LEAST_ONE);
Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE);
sessionsStore.addNewSubscription(subQos1);
sessionsStore.addNewSubscription(subQos2);
@@ -490,7 +491,8 @@ public class ProtocolProcessorTest {
processor.connectionDescriptors.put("Sub B", new ConnectionDescriptor("Sub B", null, true));
//Exercise
processor.route2Subscribers(forwardPublish);
List<Subscription> topicMatchingSubscriptions = subscriptions.matches(topic);
processor.route2Subscribers(forwardPublish, topicMatchingSubscriptions);
//Verify
assertEquals(2, publishedForwarded.size());