From 9236def453095f4d38c49d750cee69f8ebca13fa Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 16 Nov 2015 23:45:47 +0100 Subject: [PATCH] Used the message store for QoS2 messages --- .../eclipse/moquette/spi/ClientSession.java | 5 +++ .../eclipse/moquette/spi/IMessagesStore.java | 4 ++- .../eclipse/moquette/spi/ISessionsStore.java | 2 ++ .../moquette/spi/impl/ProtocolProcessor.java | 31 +++++++++---------- .../spi/persistence/MapDBPersistentStore.java | 15 +++++++++ .../server/ServerIntegrationFuseTest.java | 6 ++-- .../spi/impl/MemoryStorageService.java | 16 ++++++++++ .../spi/persistence/MemorySessionStore.java | 8 ++++- 8 files changed, 65 insertions(+), 22 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java index 59a4c1d6..0a6b2db5 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java @@ -171,4 +171,9 @@ public class ClientSession { public void enqueueToDeliver(String guid) { this.m_sessionsStore.bindToDeliver(guid, this.clientID); } + + public PublishEvent storedMessage(int messageID) { + final String guid = m_sessionsStore.mapToGuid(clientID, messageID); + return messagesStore.getMessageByGuid(guid); + } } \ No newline at end of file diff --git a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java index fd36080b..0c3e530b 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -83,8 +83,10 @@ public interface IMessagesStore { void dropMessagesInSession(String clientID); + PublishEvent getMessageByGuid(String guid); + /** - * Return the next valid packetIdentifer for the given client session. + * Return the next valid packetIdentifier for the given client session. * */ int nextPacketID(String clientID); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java index 05959d4b..1f953d01 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java @@ -91,4 +91,6 @@ public interface ISessionsStore { void secondPhaseAcknowledged(String clientID, int messageID); void secondPhaseAckWaiting(String clientID, int messageID); + + String mapToGuid(String clientID, int messageID); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index d67770e6..d548c418 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -121,7 +121,6 @@ public class ProtocolProcessor { ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) { - //m_clientIDs = clientIDs; this.m_clientIDs = new ConcurrentHashMap<>(); this.m_interceptor = interceptor; this.subscriptions = subscriptions; @@ -274,16 +273,10 @@ public class ProtocolProcessor { final String topic = msg.getTopicName(); //check if the topic can be wrote String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); - if (m_authorizator.canWrite(topic, user, clientID)) { - executePublish(clientID, msg); - m_interceptor.notifyTopicPublished(msg, clientID); - } else { + if (!m_authorizator.canWrite(topic, user, clientID)) { LOG.debug("topic {} doesn't have write credentials", topic); + return; } - } - - private void executePublish(String clientID, PublishMessage msg) { - final String topic = msg.getTopicName(); final AbstractMessage.QOSType qos = msg.getQos(); final ByteBuffer message = msg.getPayload(); boolean retain = msg.isRetainFlag(); @@ -298,9 +291,11 @@ public class ProtocolProcessor { sendPubAck(new PubAckEvent(messageID, clientID)); LOG.debug("replying with PubAck to MSG ID {}", messageID); } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 - String publishKey = String.format("%s%d", clientID, messageID); +// String publishKey = String.format("%s%d", clientID, messageID); //store the message in temp store - m_messagesStore.persistQoS2Message(publishKey, publishEvt); +//TODO use the global message store +// m_messagesStore.persistQoS2Message(publishKey, publishEvt); + m_messagesStore.storePublishForFuture(publishEvt); sendPubRec(clientID, messageID); //Next the client will send us a pub rel //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher @@ -308,14 +303,15 @@ public class ProtocolProcessor { if (retain) { if (qos == AbstractMessage.QOSType.MOST_ONE) { - //QoS == 0 && retain => clean old retained + //QoS == 0 && retain => clean old retained m_messagesStore.cleanRetained(topic); } else { m_messagesStore.storeRetained(topic, message, qos); } } + m_interceptor.notifyTopicPublished(msg, clientID); } - + /** * Specialized version to publish will testament message. */ @@ -460,10 +456,13 @@ public class ProtocolProcessor { String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); - String publishKey = String.format("%s%d", clientID, messageID); - PublishEvent evt = m_messagesStore.retrieveQoS2Message(publishKey); +// String publishKey = String.format("%s%d", clientID, messageID); +// PublishEvent evt = m_messagesStore.retrieveQoS2Message(publishKey); + //get guid, get message + ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); + PublishEvent evt = targetSession.storedMessage(messageID); forward2Subscribers(evt); - m_messagesStore.removeQoS2Message(publishKey); +// m_messagesStore.removeQoS2Message(publishKey); if (evt.isRetain()) { final String topic = evt.getTopic(); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index 28c228d0..ad02e916 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -170,6 +170,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { String guid = UUID.randomUUID().toString(); evt.setGuid(guid); m_persistentMessageStore.put(guid, convertToStored(evt)); + ConcurrentMap messageIdToGuid = m_db.getHashMap("guidsMapping" + evt.getClientID()); + messageIdToGuid.put(evt.getMessageID(), guid); return guid; } @@ -182,10 +184,23 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { return ret; } + @Override public void dropMessagesInSession(String clientID) { + m_db.getHashMap("guidsMapping" + clientID).clear(); m_persistentMessageStore.remove(clientID); } + @Override + public PublishEvent getMessageByGuid(String guid) { + return convertFromStored(m_persistentMessageStore.get(guid)); + } + + @Override + public String mapToGuid(String clientID, int messageID) { + ConcurrentMap messageIdToGuid = m_db.getHashMap("guidsMapping" + clientID); + return messageIdToGuid.get(messageID); + } + /** * Return the next valid packetIdentifier for the given client session. * */ diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java index f670d683..a50e6389 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java @@ -24,12 +24,10 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.File; import java.io.IOException; import java.util.Properties; import java.util.concurrent.TimeUnit; -import static org.eclipse.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME; import static org.junit.Assert.*; /** @@ -208,8 +206,8 @@ public class ServerIntegrationFuseTest { * subscriber connects again and receive "hello1" "hello2" */ @Test - public void checkQoS2SuscriberDisconnectReceivePersistedPublishes() throws Exception { - LOG.info("*** checkQoS2SuscriberDisconnectReceivePersistedPublishes ***"); + public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws Exception { + LOG.info("*** checkQoS2SubscriberDisconnectReceivePersistedPublishes ***"); m_mqtt.setHost("localhost", 1883); m_mqtt.setCleanSession(false); m_mqtt.setClientId("Subscriber"); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index b606d03f..70e7c35a 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -47,6 +47,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { private Map> m_secondPhaseStore = new HashMap<>(); private Map> m_inflightIDs = new HashMap<>(); private Map m_qos2Store = new HashMap<>(); + private Map> m_messageToGuids = new HashMap<>(); private MemorySessionStore m_sessionsStore; private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class); @@ -95,6 +96,9 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { String guid = UUID.randomUUID().toString(); evt.setGuid(guid); m_persistentMessageStore.put(guid, evt); + HashMap guids = (HashMap) defaultGet(m_messageToGuids, + evt.getClientID(), new HashMap()); + guids.put(evt.getMessageID(), guid); return guid; } @@ -112,6 +116,18 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { m_persistentMessageStore.remove(clientID); } + @Override + public PublishEvent getMessageByGuid(String guid) { + return m_persistentMessageStore.get(guid); + } + + @Override + public String mapToGuid(String clientID, int messageID) { + HashMap guids = (HashMap) defaultGet(m_messageToGuids, + clientID, new HashMap()); + return guids.get(messageID); + } + /** * Return the next valid packetIdentifier for the given client session. * */ diff --git a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java index e77adc72..97a40f05 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java @@ -18,6 +18,7 @@ package org.eclipse.moquette.spi.persistence; import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; +import org.eclipse.moquette.spi.impl.MemoryStorageService; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import static org.eclipse.moquette.spi.impl.Utils.defaultGet; @@ -29,7 +30,7 @@ import java.util.*; /** * - * @author andream_messagesStore + * @author andrea */ public class MemorySessionStore implements ISessionsStore { private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); @@ -208,4 +209,9 @@ public class MemorySessionStore implements ISessionsStore { messageIDs.add(messageID); m_secondPhaseStore.put(clientID, messageIDs); } + + @Override + public String mapToGuid(String clientID, int messageID) { + return ((MemoryStorageService) m_messagesStore).mapToGuid(clientID, messageID); + } }