From 9f9d01ec581e426187a93851706154f435b45bfc Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 18 Nov 2015 22:19:43 +0100 Subject: [PATCH] Changed retained persistence to use the global message store and guids, instead of duplicate msg contents --- .../eclipse/moquette/spi/ClientSession.java | 5 +- .../eclipse/moquette/spi/IMessagesStore.java | 61 ++++++++++- .../moquette/spi/impl/ProtocolProcessor.java | 60 +++++++--- .../spi/impl/events/PublishEvent.java | 103 ------------------ .../spi/impl/storage/StoredPublishEvent.java | 79 -------------- .../spi/persistence/MapDBPersistentStore.java | 69 +++++------- .../spi/impl/MemoryStorageService.java | 32 ++---- .../spi/impl/ProtocolProcessorTest.java | 24 ++-- 8 files changed, 155 insertions(+), 278 deletions(-) delete mode 100644 broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java delete mode 100644 broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java 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 0a6b2db5..4f2a0b8c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java @@ -16,7 +16,6 @@ package org.eclipse.moquette.spi; import org.eclipse.moquette.proto.messages.AbstractMessage; -import org.eclipse.moquette.spi.impl.events.PublishEvent; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; import org.slf4j.Logger; @@ -73,7 +72,7 @@ public class ClientSession { /** * @return the list of messages to be delivered for client related to the session. * */ - public List storedMessages() { + public List storedMessages() { //read all messages from enqueued store Collection guids = this.m_sessionsStore.enqueued(clientID); return messagesStore.listMessagesInSession(guids); @@ -172,7 +171,7 @@ public class ClientSession { this.m_sessionsStore.bindToDeliver(guid, this.clientID); } - public PublishEvent storedMessage(int messageID) { + public IMessagesStore.StoredMessage storedMessage(int messageID) { final String guid = m_sessionsStore.mapToGuid(clientID, messageID); return messagesStore.getMessageByGuid(guid); } 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 2f32c433..5d554868 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -17,7 +17,6 @@ package org.eclipse.moquette.spi; import java.io.Serializable; import java.nio.ByteBuffer; -import org.eclipse.moquette.spi.impl.events.PublishEvent; import org.eclipse.moquette.proto.messages.AbstractMessage; import java.util.Collection; @@ -32,6 +31,11 @@ public interface IMessagesStore { final AbstractMessage.QOSType m_qos; final byte[] m_payload; final String m_topic; + private boolean m_retained; + private String m_clientID; + //Optional attribute, available only fo QoS 1 and 2 + private Integer m_msgID; + private String m_guid; public StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) { m_qos = qos; @@ -50,6 +54,53 @@ public interface IMessagesStore { public String getTopic() { return m_topic; } + + public void setGuid(String guid) { + this.m_guid = guid; + } + + public String getGuid() { + return m_guid; + } + + public String getClientID() { + return m_clientID; + } + + public void setClientID(String m_clientID) { + this.m_clientID = m_clientID; + } + + public void setMessageID(Integer messageID) { + this.m_msgID = messageID; + } + + public Integer getMessageID() { + return m_msgID; + } + + public ByteBuffer getMessage() { + return ByteBuffer.wrap(m_payload); + } + + public void setRetained(boolean retained) { + this.m_retained = retained; + } + + public boolean isRetained() { + return m_retained; + } + + @Override + public String toString() { + return "PublishEvent{" + + "m_msgID=" + m_msgID + + ", clientID='" + m_clientID + '\'' + + ", m_retain=" + m_retained + + ", m_qos=" + m_qos + + ", m_topic='" + m_topic + '\'' + + '}'; + } } /** @@ -61,7 +112,7 @@ public interface IMessagesStore { * Persist the message. * If the message is empty then the topic is cleaned, else it's stored. */ - void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos); + void storeRetained(String topic, String guid); /** * Return a list of retained messages that satisfy the condition. @@ -72,18 +123,18 @@ public interface IMessagesStore { * Persist the message. * @return the unique id in the storage (guid). * */ - String storePublishForFuture(PublishEvent evt); + String storePublishForFuture(StoredMessage evt); /** * Return the list of persisted publishes for the given clientID. * For QoS1 and QoS2 with clean session flag, this method return the list of * missed publish events while the client was disconnected. */ - List listMessagesInSession(Collection guids); + List listMessagesInSession(Collection guids); void dropMessagesInSession(String clientID); - PublishEvent getMessageByGuid(String guid); + StoredMessage getMessageByGuid(String guid); /** * Return the next valid packetIdentifier for the given client session. 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 af9be6de..c79533c2 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 @@ -23,8 +23,8 @@ import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; +import org.eclipse.moquette.spi.IMessagesStore.*; import org.eclipse.moquette.spi.ISessionsStore; -import org.eclipse.moquette.spi.impl.events.*; import org.eclipse.moquette.spi.impl.security.IAuthorizator; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -244,14 +244,14 @@ public class ProtocolProcessor { * */ private void republishStoredInSession(ClientSession clientSession) { LOG.trace("republishStoredInSession for client <{}>", clientSession); - List publishedEvents = clientSession.storedMessages(); + List publishedEvents = clientSession.storedMessages(); if (publishedEvents.isEmpty()) { LOG.info("No stored messages for client <{}>", clientSession.clientID); return; } LOG.info("republishing stored messages to client <{}>", clientSession.clientID); - for (PublishEvent pubEvt : publishedEvents) { + for (StoredMessage pubEvt : publishedEvents) { //TODO put in flight zone directSend(clientSession, pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID()); @@ -266,6 +266,19 @@ public class ProtocolProcessor { ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); targetSession.inFlightAcknowledged(messageID); } + + private static StoredMessage asStoredMessage(PublishMessage msg) { + StoredMessage stored = new StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName()); + stored.setRetained(msg.isRetainFlag()); + stored.setMessageID(msg.getMessageID()); + return stored; + } + + private static StoredMessage asStoredMessage(WillMessage will) { + StoredMessage pub = new StoredMessage(will.getPayload().array(), will.getQos(), will.getTopic()); + pub.setRetained(will.isRetained()); + return pub; + } public void processPublish(ServerChannel session, PublishMessage msg) { LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg); @@ -278,20 +291,21 @@ public class ProtocolProcessor { return; } final AbstractMessage.QOSType qos = msg.getQos(); - final ByteBuffer message = msg.getPayload(); boolean retain = msg.isRetainFlag(); final Integer messageID = msg.getMessageID(); LOG.info("PUBLISH from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos); - PublishEvent publishEvt = new PublishEvent(clientID, msg); + String guid = null; + StoredMessage toStoreMsg = asStoredMessage(msg); + toStoreMsg.setClientID(clientID); if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0 - forward2Subscribers(publishEvt); + forward2Subscribers(toStoreMsg); } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1 - forward2Subscribers(publishEvt); + forward2Subscribers(toStoreMsg); sendPubAck(clientID, messageID); LOG.debug("replying with PubAck to MSG ID {}", messageID); } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 - m_messagesStore.storePublishForFuture(publishEvt); + guid = m_messagesStore.storePublishForFuture(toStoreMsg); sendPubRec(clientID, messageID); //Next the client will send us a pub rel //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher @@ -302,7 +316,15 @@ public class ProtocolProcessor { //QoS == 0 && retain => clean old retained m_messagesStore.cleanRetained(topic); } else { - m_messagesStore.storeRetained(topic, message, qos); + if (!msg.getPayload().hasRemaining()) { + m_messagesStore.cleanRetained(topic); + } else { + if (guid == null) { + //before wasn't stored + guid = m_messagesStore.storePublishForFuture(toStoreMsg); + } + m_messagesStore.storeRetained(topic, guid); + } } } m_interceptor.notifyTopicPublished(msg, clientID); @@ -319,9 +341,10 @@ public class ProtocolProcessor { messageId = m_messagesStore.nextPacketID(clientID); } - PublishEvent pub = new PublishEvent(will.getTopic(), will.getQos(), will.getPayload(), will.isRetained(), - clientID, messageId); - forward2Subscribers(pub); + StoredMessage tobeStored = asStoredMessage(will); + tobeStored.setClientID(clientID); + tobeStored.setMessageID(messageId); + forward2Subscribers(tobeStored); } @@ -329,7 +352,7 @@ public class ProtocolProcessor { * 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 forward2Subscribers(PublishEvent pubEvt) { + void forward2Subscribers(StoredMessage pubEvt) { final String topic = pubEvt.getTopic(); final AbstractMessage.QOSType publishingQos = pubEvt.getQos(); final ByteBuffer origMessage = pubEvt.getMessage(); @@ -452,13 +475,16 @@ public class ProtocolProcessor { int messageID = msg.getMessageID(); LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); - PublishEvent evt = targetSession.storedMessage(messageID); + StoredMessage evt = targetSession.storedMessage(messageID); forward2Subscribers(evt); - if (evt.isRetain()) { + if (evt.isRetained()) { final String topic = evt.getTopic(); - final AbstractMessage.QOSType qos = evt.getQos(); - m_messagesStore.storeRetained(topic, evt.getMessage(), qos); + if (!evt.getMessage().hasRemaining()) { + m_messagesStore.cleanRetained(topic); + } else { + m_messagesStore.storeRetained(topic, evt.getGuid()); + } } sendPubComp(clientID, messageID); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java deleted file mode 100644 index 410b1979..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java +++ /dev/null @@ -1,103 +0,0 @@ -/* - * Copyright (c) 2012-2015 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ -package org.eclipse.moquette.spi.impl.events; - -import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType; -import org.eclipse.moquette.proto.messages.PublishMessage; - -import java.nio.ByteBuffer; - -/** - * - * @author andrea - */ -public class PublishEvent { - String m_topic; - QOSType m_qos; - //byte[] m_message; - ByteBuffer m_message; - boolean m_retain; - String m_clientID; - //Optional attribute, available only fo QoS 1 and 2 - Integer m_msgID; - private String guid; - - public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain, - String clientID, Integer msgID) { - m_topic = topic; - m_qos = qos; - m_message = message; - m_retain = retain; - m_clientID = clientID; - if (qos != QOSType.MOST_ONE) { - m_msgID = msgID; - } - } - - public PublishEvent(String clientID, PublishMessage msg) { - m_clientID = clientID; - m_topic = msg.getTopicName(); - m_qos = msg.getQos(); - m_message = msg.getPayload(); - m_retain = msg.isRetainFlag(); - if (msg.getQos() != QOSType.MOST_ONE) { - m_msgID = msg.getMessageID(); - } - } - - public String getTopic() { - return m_topic; - } - - public QOSType getQos() { - return m_qos; - } - - public ByteBuffer getMessage() { - return m_message; - } - - public boolean isRetain() { - return m_retain; - } - - public String getClientID() { - return m_clientID; - } - - public Integer getMessageID() { - return m_msgID; - } - - public void setGuid(String guid) { - this.guid = guid; - } - - public String getGuid() { - return guid; - } - - @Override - public String toString() { - return "PublishEvent{" + - "m_msgID=" + m_msgID + - ", clientID='" + m_clientID + '\'' + - ", m_retain=" + m_retain + - ", m_qos=" + m_qos + - ", m_topic='" + m_topic + '\'' + - '}'; - } -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java deleted file mode 100644 index b9613911..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/storage/StoredPublishEvent.java +++ /dev/null @@ -1,79 +0,0 @@ -/* - * Copyright (c) 2012-2015 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ -package org.eclipse.moquette.spi.impl.storage; - -import java.io.Serializable; -import java.nio.ByteBuffer; -import org.eclipse.moquette.spi.impl.events.PublishEvent; -import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType; - -/** - * Publish event serialized to the DB. - * - * @author andrea - */ -public class StoredPublishEvent implements Serializable { - String m_topic; - QOSType m_qos; - byte[] m_message; - boolean m_retain; - String m_clientID; - //Optional attribute, available only fo QoS 1 and 2 - Integer m_msgID; - String m_guid; - - public StoredPublishEvent(PublishEvent wrapped) { - m_topic = wrapped.getTopic(); - m_qos = wrapped.getQos(); - m_retain = wrapped.isRetain(); - m_clientID = wrapped.getClientID(); - m_msgID = wrapped.getMessageID(); - - ByteBuffer buffer = wrapped.getMessage(); - m_message = new byte[buffer.remaining()]; - buffer.get(m_message); - buffer.rewind(); - m_guid = wrapped.getGuid(); - } - - public String getTopic() { - return m_topic; - } - - public QOSType getQos() { - return m_qos; - } - - public byte[] getMessage() { - return m_message; - } - - public boolean isRetain() { - return m_retain; - } - - public String getClientID() { - return m_clientID; - } - - public Integer getMessageID() { - return m_msgID; - } - - public String getGuid() { - return m_guid; - } -} 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 b38c3d3e..59581ee6 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 @@ -17,13 +17,10 @@ package org.eclipse.moquette.spi.persistence; import org.eclipse.moquette.proto.MQTTException; -import org.eclipse.moquette.proto.messages.AbstractMessage; import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; -import org.eclipse.moquette.spi.impl.events.PublishEvent; -import org.eclipse.moquette.spi.impl.storage.StoredPublishEvent; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.mapdb.DB; import org.mapdb.DBMaker; @@ -33,7 +30,6 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.io.Serializable; -import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executors; @@ -63,9 +59,10 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class); - private ConcurrentMap m_retainedStore; + //maps clientID -> guid + private ConcurrentMap m_retainedStore; //maps guid to message, it's message store - private ConcurrentMap m_persistentMessageStore; + private ConcurrentMap m_persistentMessageStore; //maps clientID->[MessageId -> guid] private ConcurrentMap> m_inflightStore; //map clientID <-> set of currently in flight packet identifiers @@ -87,7 +84,6 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { /* * The default constructor will create an in memory store as no file path was specified */ - public MapDBPersistentStore() { this.m_storePath = ""; } @@ -134,16 +130,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } @Override - public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) { - if (!message.hasRemaining()) { - //clean the message from topic - m_retainedStore.remove(topic); - } else { - //store the message to the topic - byte[] raw = new byte[message.remaining()]; - message.get(raw); - m_retainedStore.put(topic, new StoredMessage(raw, qos, topic)); - } + public void storeRetained(String topic, String guid) { + m_retainedStore.put(topic, guid); } @Override @@ -151,8 +139,9 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); List results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - StoredMessage storedMsg = entry.getValue(); + for (Map.Entry entry : m_retainedStore.entrySet()) { + final String guid = entry.getValue(); + StoredMessage storedMsg = m_persistentMessageStore.get(guid); if (condition.match(entry.getKey())) { results.add(storedMsg); } @@ -162,42 +151,50 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } @Override - public String storePublishForFuture(PublishEvent evt) { + public String storePublishForFuture(StoredMessage evt) { LOG.debug("storePublishForFuture store evt {}", evt); + if (evt.getClientID() == null) { + LOG.error("persisting a message without a clientID, bad programming error msg: {}", evt); + throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error"); + } String guid = UUID.randomUUID().toString(); evt.setGuid(guid); - m_persistentMessageStore.put(guid, convertToStored(evt)); - ConcurrentMap messageIdToGuid = m_db.getHashMap("guidsMapping" + evt.getClientID()); + m_persistentMessageStore.put(guid, evt); + ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(evt.getClientID())); messageIdToGuid.put(evt.getMessageID(), guid); return guid; } @Override - public List listMessagesInSession(Collection guids) { - List ret = new ArrayList<>(); + public List listMessagesInSession(Collection guids) { + List ret = new ArrayList<>(); for (String guid : guids) { - ret.add(convertFromStored(m_persistentMessageStore.get(guid))) ; + ret.add(m_persistentMessageStore.get(guid)); } return ret; } @Override public void dropMessagesInSession(String clientID) { - m_db.getHashMap("guidsMapping" + clientID).clear(); + m_db.getHashMap(messageId2GuidsMapName(clientID)).clear(); m_persistentMessageStore.remove(clientID); } @Override - public PublishEvent getMessageByGuid(String guid) { - return convertFromStored(m_persistentMessageStore.get(guid)); + public StoredMessage getMessageByGuid(String guid) { + return m_persistentMessageStore.get(guid); } @Override public String mapToGuid(String clientID, int messageID) { - ConcurrentMap messageIdToGuid = m_db.getHashMap("guidsMapping" + clientID); + ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); return messageIdToGuid.get(messageID); } + private String messageId2GuidsMapName(String clientID) { + return "guidsMapping_" + clientID; + } + /** * Return the next valid packetIdentifier for the given client session. * */ @@ -413,18 +410,4 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { this.m_scheduler.shutdown(); LOG.debug("Persistence commit scheduler is shutdown"); } - - private StoredPublishEvent convertToStored(PublishEvent evt) { - return new StoredPublishEvent(evt); - } - - private PublishEvent convertFromStored(StoredPublishEvent evt) { - byte[] message = evt.getMessage(); - ByteBuffer bbmessage = ByteBuffer.wrap(message); - //bbmessage.flip(); - PublishEvent pubEvt = new PublishEvent(evt.getTopic(), evt.getQos(), - bbmessage, evt.isRetain(), evt.getClientID(), evt.getMessageID()); - pubEvt.setGuid(evt.getGuid()); - return pubEvt; - } } 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 63a552f2..0a38e961 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 @@ -15,16 +15,13 @@ */ package org.eclipse.moquette.spi.impl; -import java.nio.ByteBuffer; import java.util.*; import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; -import org.eclipse.moquette.spi.impl.events.PublishEvent; import org.eclipse.moquette.spi.persistence.MemorySessionStore; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; -import org.eclipse.moquette.proto.messages.AbstractMessage; import org.eclipse.moquette.spi.ISessionsStore; import org.slf4j.Logger; @@ -36,9 +33,9 @@ import static org.eclipse.moquette.spi.impl.Utils.defaultGet; */ public class MemoryStorageService implements IMessagesStore, ISessionsStore { - private Map m_retainedStore = new HashMap<>(); + private Map m_retainedStore = new HashMap<>(); //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map) - private Map m_persistentMessageStore = new HashMap<>(); + private Map m_persistentMessageStore = new HashMap<>(); //maps clientID->[MessageId -> guid] private Map> m_inflightStore = new HashMap<>(); //maps clientID->[guid*] @@ -61,16 +58,8 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public void storeRetained(String topic, ByteBuffer message, AbstractMessage.QOSType qos) { - if (!message.hasRemaining()) { - //clean the message from topic - m_retainedStore.remove(topic); - } else { - //store the message to the topic - byte[] raw = new byte[message.remaining()]; - message.get(raw); - m_retainedStore.put(topic, new StoredMessage(raw, qos, topic)); - } + public void storeRetained(String topic, String guid) { + m_retainedStore.put(topic, guid); } @Override @@ -79,8 +68,9 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { List results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - StoredMessage storedMsg = entry.getValue(); + for (Map.Entry entry : m_retainedStore.entrySet()) { + final String guid = entry.getValue(); + StoredMessage storedMsg = m_persistentMessageStore.get(guid); if (condition.match(entry.getKey())) { results.add(storedMsg); } @@ -90,7 +80,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public String storePublishForFuture(PublishEvent evt) { + public String storePublishForFuture(StoredMessage evt) { LOG.debug("storePublishForFuture store evt {}", evt); String guid = UUID.randomUUID().toString(); evt.setGuid(guid); @@ -102,8 +92,8 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public List listMessagesInSession(Collection guids) { - List ret = new ArrayList<>(); + public List listMessagesInSession(Collection guids) { + List ret = new ArrayList<>(); for (String guid : guids) { ret.add(m_persistentMessageStore.get(guid)); } @@ -116,7 +106,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public PublishEvent getMessageByGuid(String guid) { + public StoredMessage getMessageByGuid(String guid) { return m_persistentMessageStore.get(guid); } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 6006ffd1..5a4c5f34 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -26,14 +26,15 @@ import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; +import org.eclipse.moquette.spi.IMessagesStore.*; import org.eclipse.moquette.spi.ISessionsStore; -import org.eclipse.moquette.spi.impl.events.PublishEvent; import org.eclipse.moquette.spi.persistence.MemorySessionStore; import org.eclipse.moquette.spi.impl.security.PermitAllAuthorizator; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType; + import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; @@ -357,8 +358,11 @@ public class ProtocolProcessorTest { SubscriptionsStore subs = mock(SubscriptionsStore.class); List emptySubs = Collections.emptyList(); when(subs.matches(anyString())).thenReturn(emptySubs); - PublishEvent retainedMessage = new PublishEvent("/topic", QOSType.EXACTLY_ONCE, - ByteBuffer.wrap("Hello".getBytes()), true, FAKE_PUBLISHER_ID, 120); + + StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(), QOSType.EXACTLY_ONCE, "/topic"); + retainedMessage.setRetained(true); + retainedMessage.setMessageID(120); + retainedMessage.setClientID(FAKE_PUBLISHER_ID); m_storageService.storePublishForFuture(retainedMessage); m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); @@ -464,12 +468,14 @@ public class ProtocolProcessorTest { assertTrue(messages.isEmpty()); } - List publishedForwarded = new ArrayList<>(); + List publishedForwarded = new ArrayList<>(); @Test public void testForwardPublishWithCorrectQos() { - ByteBuffer payload = ByteBuffer.wrap("Hello world MQTT!!".getBytes()); - PublishEvent forwardPublish = new PublishEvent("a/b", QOSType.EXACTLY_ONCE, payload, true, "Publisher", 1); + StoredMessage forwardPublish = new StoredMessage("Hello world MQTT!!".getBytes(), QOSType.EXACTLY_ONCE, "a/b"); + forwardPublish.setRetained(true); + forwardPublish.setMessageID(1); + IMessagesStore memoryMessageStore = new MemoryStorageService(); ISessionsStore sessionsStore = new MemorySessionStore(memoryMessageStore); sessionsStore.createNewSession("Sub A", false).activate(); @@ -487,7 +493,11 @@ public class ProtocolProcessorTest { @Override protected void directSend(ClientSession session, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) { - publishedForwarded.add(new PublishEvent(topic, qos, message, retained, session.clientID, messageID)); + StoredMessage msgToStore = new StoredMessage(message.array(), qos, topic); + msgToStore.setRetained(retained); + msgToStore.setMessageID(messageID); + msgToStore.setClientID(session.clientID); + publishedForwarded.add(msgToStore); } }; processor.init(subscriptions, memoryMessageStore, sessionsStore, null, true, null, NO_OBSERVERS_INTERCEPTOR);