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

Minor change, just renames

Browse Source
This commit is contained in:
andsel
2015-12-08 15:15:51 +01:00
parent 8c88567c78
commit 6805cf89ee
2 changed files with 12 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+11 -13
View File
@@ -291,7 +291,6 @@ public class ProtocolProcessor {
return; return;
} }
final AbstractMessage.QOSType qos = msg.getQos(); final AbstractMessage.QOSType qos = msg.getQos();
boolean retain = msg.isRetainFlag();
final Integer messageID = msg.getMessageID(); final Integer messageID = msg.getMessageID();
LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos); LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos);
@@ -299,9 +298,9 @@ public class ProtocolProcessor {
IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg);
toStoreMsg.setClientID(clientID); toStoreMsg.setClientID(clientID);
if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0 if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0
forward2Subscribers(toStoreMsg); route2Subscribers(toStoreMsg);
} else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1 } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1
forward2Subscribers(toStoreMsg); route2Subscribers(toStoreMsg);
sendPubAck(clientID, messageID); sendPubAck(clientID, messageID);
LOG.debug("replying with PubAck to MSG ID {}", messageID); LOG.debug("replying with PubAck to MSG ID {}", messageID);
} else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
@@ -311,7 +310,7 @@ public class ProtocolProcessor {
//NB publish to subscribers for QoS 2 happen upon PUBREL from publisher //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
} }
if (retain) { if (msg.isRetainFlag()) {
if (qos == AbstractMessage.QOSType.MOST_ONE) { if (qos == AbstractMessage.QOSType.MOST_ONE) {
//QoS == 0 && retain => clean old retained //QoS == 0 && retain => clean old retained
m_messagesStore.cleanRetained(topic); m_messagesStore.cleanRetained(topic);
@@ -344,19 +343,18 @@ public class ProtocolProcessor {
IMessagesStore.StoredMessage tobeStored = asStoredMessage(will); IMessagesStore.StoredMessage tobeStored = asStoredMessage(will);
tobeStored.setClientID(clientID); tobeStored.setClientID(clientID);
tobeStored.setMessageID(messageId); tobeStored.setMessageID(messageId);
forward2Subscribers(tobeStored); route2Subscribers(tobeStored);
} }
/** /**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2 * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */ * */
//TODO rename to route2Subscribers void route2Subscribers(IMessagesStore.StoredMessage pubMsg) {
void forward2Subscribers(IMessagesStore.StoredMessage pubEvt) { final String topic = pubMsg.getTopic();
final String topic = pubEvt.getTopic(); final AbstractMessage.QOSType publishingQos = pubMsg.getQos();
final AbstractMessage.QOSType publishingQos = pubEvt.getQos(); final ByteBuffer origMessage = pubMsg.getMessage();
final ByteBuffer origMessage = pubEvt.getMessage(); LOG.debug("route2Subscribers republishing to existing subscribers that matches the topic {}", topic);
LOG.debug("forward2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isTraceEnabled()) { if (LOG.isTraceEnabled()) {
LOG.trace("content <{}>", DebugUtils.payload2Str(origMessage)); LOG.trace("content <{}>", DebugUtils.payload2Str(origMessage));
LOG.trace("subscription tree {}", subscriptions.dumpTree()); LOG.trace("subscription tree {}", subscriptions.dumpTree());
@@ -364,7 +362,7 @@ public class ProtocolProcessor {
//if QoS 1 or 2 store the message //if QoS 1 or 2 store the message
String guid = null; String guid = null;
if (publishingQos == QOSType.EXACTLY_ONCE || publishingQos == QOSType.LEAST_ONE) { if (publishingQos == QOSType.EXACTLY_ONCE || publishingQos == QOSType.LEAST_ONE) {
guid = m_messagesStore.storePublishForFuture(pubEvt); guid = m_messagesStore.storePublishForFuture(pubMsg);
} }
for (final Subscription sub : subscriptions.matches(topic)) { for (final Subscription sub : subscriptions.matches(topic)) {
@@ -476,7 +474,7 @@ public class ProtocolProcessor {
LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID); IMessagesStore.StoredMessage evt = targetSession.storedMessage(messageID);
forward2Subscribers(evt); route2Subscribers(evt);
if (evt.isRetained()) { if (evt.isRetained()) {
final String topic = evt.getTopic(); final String topic = evt.getTopic();
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+1 -2
View File
@@ -24,7 +24,6 @@ import io.moquette.spi.IMatchingCondition;
import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore;
import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore; import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.security.PermitAllAuthorizator;
import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore; import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
@@ -507,7 +506,7 @@ public class ProtocolProcessorTest {
processor.init(subscriptions, memoryMessageStore, sessionsStore, null, true, null, NO_OBSERVERS_INTERCEPTOR); processor.init(subscriptions, memoryMessageStore, sessionsStore, null, true, null, NO_OBSERVERS_INTERCEPTOR);
//Exercise //Exercise
processor.forward2Subscribers(forwardPublish); processor.route2Subscribers(forwardPublish);
//Verify //Verify
assertEquals(2, publishedForwarded.size()); assertEquals(2, publishedForwarded.size());