diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index ca90217c..91f5c215 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -78,14 +78,14 @@ public class ClientSession { * */ public List storedMessages() { //read all messages from enqueued store - Collection guids = this.m_sessionsStore.enqueued(clientID); + Collection guids = this.m_sessionsStore.enqueued(clientID); return messagesStore.listMessagesInSession(guids); } /** * Remove the messages stored in a cleanSession false. */ - public void removeEnqueued(String guid) { + public void removeEnqueued(MessageGUID guid) { this.m_sessionsStore.removeEnqueued(this.clientID, guid); } @@ -175,22 +175,22 @@ public class ClientSession { m_sessionsStore.inFlightAck(this.clientID, messageID); } - public void inFlightAckWaiting(String guid, int messageID) { + public void inFlightAckWaiting(MessageGUID guid, int messageID) { LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid); m_sessionsStore.inFlight(this.clientID, messageID, guid); } public IMessagesStore.StoredMessage secondPhaseAcknowledged(int messageID) { - String guid = m_sessionsStore.secondPhaseAcknowledged(clientID, messageID); + MessageGUID guid = m_sessionsStore.secondPhaseAcknowledged(clientID, messageID); return messagesStore.getMessageByGuid(guid); } - public void enqueueToDeliver(String guid) { + public void enqueueToDeliver(MessageGUID guid) { this.m_sessionsStore.bindToDeliver(guid, this.clientID); } public IMessagesStore.StoredMessage storedMessage(int messageID) { - final String guid = m_sessionsStore.mapToGuid(clientID, messageID); + final MessageGUID guid = m_sessionsStore.mapToGuid(clientID, messageID); return messagesStore.getMessageByGuid(guid); } diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index f2ad8ce1..e2994d0d 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -35,7 +35,7 @@ public interface IMessagesStore { private String m_clientID; //Optional attribute, available only fo QoS 1 and 2 private Integer m_msgID; - private String m_guid; + private MessageGUID m_guid; private int referenceCounter = 0; public StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) { @@ -56,11 +56,11 @@ public interface IMessagesStore { return m_topic; } - public void setGuid(String guid) { + public void setGuid(MessageGUID guid) { this.m_guid = guid; } - public String getGuid() { + public MessageGUID getGuid() { return m_guid; } @@ -125,7 +125,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, String guid); + void storeRetained(String topic, MessageGUID guid); /** * Return a list of retained messages that satisfy the condition. @@ -136,22 +136,22 @@ public interface IMessagesStore { * Persist the message. * @return the unique id in the storage (guid). * */ - String storePublishForFuture(StoredMessage evt); + MessageGUID 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); - StoredMessage getMessageByGuid(String guid); + StoredMessage getMessageByGuid(MessageGUID guid); void cleanRetained(String topic); - void incUsageCounter(String guid); + void incUsageCounter(MessageGUID guid); - void decUsageCounter(String guid); + void decUsageCounter(MessageGUID guid); } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index e1237b01..fc7506b5 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -112,7 +112,7 @@ public interface ISessionsStore { /** * Save the binding messageID, clientID <-> guid * */ - void inFlight(String clientID, int messageID, String guid); + void inFlight(String clientID, int messageID, MessageGUID guid); /** * Return the next valid packetIdentifier for the given client session. @@ -122,26 +122,26 @@ public interface ISessionsStore { /** * Store the guid to be later published. * */ - void bindToDeliver(String guid, String clientID); + void bindToDeliver(MessageGUID guid, String clientID); /** * List the guids for retained messages for the session * */ - Collection enqueued(String clientID); + Collection enqueued(String clientID); /** * Remove form the queue of stored messages for session. * */ - void removeEnqueued(String clientID, String guid); + void removeEnqueued(String clientID, MessageGUID guid); void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID); /** * @return the guid of message just acked. * */ - String secondPhaseAcknowledged(String clientID, int messageID); + MessageGUID secondPhaseAcknowledged(String clientID, int messageID); - String mapToGuid(String clientID, int messageID); + MessageGUID mapToGuid(String clientID, int messageID); StoredMessage getInflightMessage(String clientID, int messageID); } diff --git a/broker/src/main/java/io/moquette/spi/MessageGUID.java b/broker/src/main/java/io/moquette/spi/MessageGUID.java new file mode 100644 index 00000000..41213238 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/MessageGUID.java @@ -0,0 +1,52 @@ +/* + * 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 io.moquette.spi; + +import java.io.Serializable; + +/** + * Value object for GUIDs of messages. + * */ +public class MessageGUID implements Serializable { + private final String guid; + + public MessageGUID(String guid) { + this.guid = guid; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + MessageGUID that = (MessageGUID) o; + + return guid != null ? guid.equals(that.guid) : that.guid == null; + + } + + @Override + public int hashCode() { + return guid != null ? guid.hashCode() : 0; + } + + @Override + public String toString() { + return "MessageGUID{" + + "guid='" + guid + '\'' + + '}'; + } +} \ No newline at end of file diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 53592df0..6320861b 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -25,11 +25,8 @@ import io.moquette.interception.InterceptHandler; import io.moquette.server.ConnectionDescriptor; import io.moquette.server.netty.AutoFlushHandler; import io.moquette.server.netty.NettyUtils; -import io.moquette.spi.ClientSession; -import io.moquette.spi.IMatchingCondition; -import io.moquette.spi.IMessagesStore; +import io.moquette.spi.*; import io.moquette.spi.IMessagesStore.StoredMessage; -import io.moquette.spi.ISessionsStore; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -150,7 +147,6 @@ public class ProtocolProcessor { * @param allowZeroByteClientId true to allow clients connect without a clientid * @param authorizator used to apply ACL policies to publishes and subscriptions. * @param interceptor to notify events to an intercept handler - * @param serverPort */ void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, @@ -387,7 +383,7 @@ public class ProtocolProcessor { final Integer messageID = msg.getMessageID(); LOG.info("PUBLISH on server {} from clientID <{}> on topic <{}> with QoS {}", m_server_port, clientID, topic, qos); - String guid = null; + MessageGUID guid = null; IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); toStoreMsg.setClientID(clientID); if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0 @@ -439,7 +435,7 @@ public class ProtocolProcessor { final String topic = msg.getTopicName(); LOG.info("embedded PUBLISH on topic <{}> with QoS {}", topic, qos); - String guid = null; + MessageGUID guid = null; IMessagesStore.StoredMessage toStoreMsg = asStoredMessage(msg); if (msg.getClientId() == null || msg.getClientId().isEmpty()) { toStoreMsg.setClientID("BROKER_SELF"); @@ -498,7 +494,7 @@ public class ProtocolProcessor { LOG.trace("subscription tree {}", subscriptions.dumpTree()); } //if QoS 1 or 2 store the message - String guid = null; + MessageGUID guid = null; if (publishingQos == QOSType.EXACTLY_ONCE || publishingQos == QOSType.LEAST_ONE) { guid = m_messagesStore.storePublishForFuture(pubMsg); } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java index 9affdc65..54668069 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -17,6 +17,7 @@ package io.moquette.spi.persistence; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; +import io.moquette.spi.MessageGUID; import org.mapdb.DB; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,9 +37,9 @@ class MapDBMessagesStore implements IMessagesStore { private DB m_db; //maps clientID -> guid - private ConcurrentMap m_retainedStore; + private ConcurrentMap m_retainedStore; //maps guid to message, it's message store - private ConcurrentMap m_persistentMessageStore; + private ConcurrentMap m_persistentMessageStore; MapDBMessagesStore(DB db) { @@ -52,7 +53,7 @@ class MapDBMessagesStore implements IMessagesStore { } @Override - public void storeRetained(String topic, String guid) { + public void storeRetained(String topic, MessageGUID guid) { m_retainedStore.put(topic, guid); } @@ -61,8 +62,8 @@ class MapDBMessagesStore implements IMessagesStore { LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); List results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - final String guid = entry.getValue(); + for (Map.Entry entry : m_retainedStore.entrySet()) { + final MessageGUID guid = entry.getValue(); StoredMessage storedMsg = m_persistentMessageStore.get(guid); if (condition.match(entry.getKey())) { results.add(storedMsg); @@ -73,25 +74,25 @@ class MapDBMessagesStore implements IMessagesStore { } @Override - public String storePublishForFuture(StoredMessage evt) { + public MessageGUID 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(); + MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); evt.setGuid(guid); LOG.debug("storePublishForFuture guid <{}>", guid); m_persistentMessageStore.put(guid, evt); - ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID())); + ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID())); messageIdToGuid.put(evt.getMessageID(), guid); return guid; } @Override - public List listMessagesInSession(Collection guids) { + public List listMessagesInSession(Collection guids) { List ret = new ArrayList<>(); - for (String guid : guids) { + for (MessageGUID guid : guids) { ret.add(m_persistentMessageStore.get(guid)); } return ret; @@ -99,14 +100,14 @@ class MapDBMessagesStore implements IMessagesStore { @Override public void dropMessagesInSession(String clientID) { - ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)); - for (String guid : messageIdToGuid.values()) { + ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)); + for (MessageGUID guid : messageIdToGuid.values()) { removeStoredMessage(guid); } messageIdToGuid.clear(); } - void removeStoredMessage(String guid) { + void removeStoredMessage(MessageGUID guid) { //remove only the not retained and no more referenced StoredMessage storedMessage = m_persistentMessageStore.get(guid); if (!storedMessage.isRetained() && storedMessage.getReferenceCounter() == 0) { @@ -116,7 +117,7 @@ class MapDBMessagesStore implements IMessagesStore { } @Override - public StoredMessage getMessageByGuid(String guid) { + public StoredMessage getMessageByGuid(MessageGUID guid) { return m_persistentMessageStore.get(guid); } @@ -126,14 +127,14 @@ class MapDBMessagesStore implements IMessagesStore { } @Override - public void incUsageCounter(String guid) { + public void incUsageCounter(MessageGUID guid) { IMessagesStore.StoredMessage storedMessage = m_persistentMessageStore.get(guid); storedMessage.incReferenceCounter(); m_persistentMessageStore.replace(guid, storedMessage); } @Override - public void decUsageCounter(String guid) { + public void decUsageCounter(MessageGUID guid) { IMessagesStore.StoredMessage storedMessage = m_persistentMessageStore.get(guid); storedMessage.decReferenceCounter(); m_persistentMessageStore.replace(guid, storedMessage); diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index 11c4180b..686665a5 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -18,6 +18,7 @@ package io.moquette.spi.persistence; import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.Utils; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; @@ -38,14 +39,14 @@ class MapDBSessionsStore implements ISessionsStore { private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class); //maps clientID->[MessageId -> guid] - private ConcurrentMap> m_inflightStore; + private ConcurrentMap> m_inflightStore; //map clientID <-> set of currently in flight packet identifiers private Map> m_inFlightIds; private ConcurrentMap m_persistentSessions; //maps clientID->[guid*], insertion order cares, it's queue - private ConcurrentMap> m_enqueuedStore; + private ConcurrentMap> m_enqueuedStore; //maps clientID->[MessageId -> guid] - private ConcurrentMap> m_secondPhaseStore; + private ConcurrentMap> m_secondPhaseStore; private final DB m_db; private final MapDBMessagesStore m_messagesStore; @@ -172,7 +173,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void inFlightAck(String clientID, int messageID) { LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); - Map m = this.m_inflightStore.get(clientID); + Map m = this.m_inflightStore.get(clientID); if (m == null) { LOG.error("Can't find the inFlight record for client <{}>", clientID); return; @@ -187,8 +188,8 @@ class MapDBSessionsStore implements ISessionsStore { } @Override - public void inFlight(String clientID, int messageID, String guid) { - Map m = this.m_inflightStore.get(clientID); + public void inFlight(String clientID, int messageID, MessageGUID guid) { + Map m = this.m_inflightStore.get(clientID); if (m == null) { m = new HashMap<>(); } @@ -198,21 +199,21 @@ class MapDBSessionsStore implements ISessionsStore { } @Override - public void bindToDeliver(String guid, String clientID) { - List guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); + public void bindToDeliver(MessageGUID guid, String clientID) { + List guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); guids.add(guid); m_enqueuedStore.put(clientID, guids); m_messagesStore.incUsageCounter(guid); } @Override - public Collection enqueued(String clientID) { - return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); + public Collection enqueued(String clientID) { + return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); } @Override - public void removeEnqueued(String clientID, String guid) { - List guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); + public void removeEnqueued(String clientID, MessageGUID guid) { + List guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList()); guids.remove(guid); m_enqueuedStore.put(clientID, guids); m_messagesStore.decUsageCounter(guid); @@ -223,12 +224,12 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); - Map m = this.m_inflightStore.get(clientID); + Map m = this.m_inflightStore.get(clientID); if (m == null) { LOG.error("Can't find the inFlight record for client <{}>", clientID); return; } - String guid = m.remove(messageID); + MessageGUID guid = m.remove(messageID); //remove from the ids store Set inFlightForClient = this.m_inFlightIds.get(clientID); @@ -237,22 +238,22 @@ class MapDBSessionsStore implements ISessionsStore { } LOG.info("Moving to second phase store"); - Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); messageIDs.put(messageID, guid); m_secondPhaseStore.put(clientID, messageIDs); } @Override - public String secondPhaseAcknowledged(String clientID, int messageID) { - Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); - String guid = messageIDs.remove(messageID); + public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); + MessageGUID guid = messageIDs.remove(messageID); m_secondPhaseStore.put(clientID, messageIDs); return guid; } @Override - public String mapToGuid(String clientID, int messageID) { - ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); + public MessageGUID mapToGuid(String clientID, int messageID) { + ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); return messageIdToGuid.get(messageID); } @@ -262,10 +263,10 @@ class MapDBSessionsStore implements ISessionsStore { @Override public StoredMessage getInflightMessage(String clientID, int messageID) { - Map clientEntries = m_inflightStore.get(clientID); + Map clientEntries = m_inflightStore.get(clientID); if (clientEntries == null) return null; - String guid = clientEntries.get(messageID); + MessageGUID guid = clientEntries.get(messageID); LOG.info("inflight messageID {} guid <{}>", messageID, guid); if (guid == null) return null; diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java index 685f62eb..75af2e00 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java @@ -17,6 +17,7 @@ package io.moquette.spi.impl; import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMatchingCondition; +import io.moquette.spi.MessageGUID; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,11 +32,11 @@ public class MemoryMessagesStore implements IMessagesStore { private static final Logger LOG = LoggerFactory.getLogger(MemoryMessagesStore.class); - private Map m_retainedStore = new HashMap<>(); - private Map m_persistentMessageStore = new HashMap<>(); - private Map> m_messageToGuids; + private Map m_retainedStore = new HashMap<>(); + private Map m_persistentMessageStore = new HashMap<>(); + private Map> m_messageToGuids; - MemoryMessagesStore(Map> messageToGuids) { + MemoryMessagesStore(Map> messageToGuids) { m_messageToGuids = messageToGuids; } @@ -44,7 +45,7 @@ public class MemoryMessagesStore implements IMessagesStore { } @Override - public void storeRetained(String topic, String guid) { + public void storeRetained(String topic, MessageGUID guid) { m_retainedStore.put(topic, guid); } @@ -54,8 +55,8 @@ public class MemoryMessagesStore implements IMessagesStore { List results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - final String guid = entry.getValue(); + for (Map.Entry entry : m_retainedStore.entrySet()) { + final MessageGUID guid = entry.getValue(); StoredMessage storedMsg = m_persistentMessageStore.get(guid); if (condition.match(entry.getKey())) { results.add(storedMsg); @@ -66,21 +67,21 @@ public class MemoryMessagesStore implements IMessagesStore { } @Override - public String storePublishForFuture(StoredMessage evt) { + public MessageGUID storePublishForFuture(StoredMessage evt) { LOG.debug("storePublishForFuture store evt {}", evt); - String guid = UUID.randomUUID().toString(); + MessageGUID guid = new MessageGUID(UUID.randomUUID().toString()); evt.setGuid(guid); m_persistentMessageStore.put(guid, evt); - HashMap guids = (HashMap) defaultGet(m_messageToGuids, - evt.getClientID(), new HashMap()); + HashMap guids = (HashMap) defaultGet(m_messageToGuids, + evt.getClientID(), new HashMap()); guids.put(evt.getMessageID(), guid); return guid; } @Override - public List listMessagesInSession(Collection guids) { + public List listMessagesInSession(Collection guids) { List ret = new ArrayList<>(); - for (String guid : guids) { + for (MessageGUID guid : guids) { ret.add(m_persistentMessageStore.get(guid)); } return ret; @@ -88,11 +89,17 @@ public class MemoryMessagesStore implements IMessagesStore { @Override public void dropMessagesInSession(String clientID) { - m_persistentMessageStore.remove(clientID); + Map messageGUIDMap = m_messageToGuids.get(clientID); + if (messageGUIDMap == null || messageGUIDMap.isEmpty()) { + return; + } + for (MessageGUID guid : messageGUIDMap.values()) { + m_persistentMessageStore.remove(guid); + } } @Override - public StoredMessage getMessageByGuid(String guid) { + public StoredMessage getMessageByGuid(MessageGUID guid) { return m_persistentMessageStore.get(guid); } @@ -102,14 +109,14 @@ public class MemoryMessagesStore implements IMessagesStore { } @Override - public void incUsageCounter(String guid) { + public void incUsageCounter(MessageGUID guid) { IMessagesStore.StoredMessage storedMessage = m_persistentMessageStore.get(guid); storedMessage.incReferenceCounter(); m_persistentMessageStore.replace(guid, storedMessage); } @Override - public void decUsageCounter(String guid) { + public void decUsageCounter(MessageGUID guid) { IMessagesStore.StoredMessage storedMessage = m_persistentMessageStore.get(guid); storedMessage.decReferenceCounter(); m_persistentMessageStore.replace(guid, storedMessage); diff --git a/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java b/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java index 54cfea1c..794695f6 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java @@ -17,6 +17,7 @@ package io.moquette.spi.impl; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.MessageGUID; import io.moquette.spi.persistence.MemorySessionStore; import java.util.HashMap; @@ -28,7 +29,7 @@ public class MemoryStorageService { private MemoryMessagesStore m_messagesStore; public void initStore() { - Map> messageToGuids = new HashMap<>(); + Map> messageToGuids = new HashMap<>(); m_messagesStore = new MemoryMessagesStore(messageToGuids); m_sessionsStore = new MemorySessionStore(m_messagesStore, messageToGuids); } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 983dece9..15707b74 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -20,11 +20,8 @@ import io.moquette.interception.InterceptHandler; import io.moquette.parser.proto.messages.AbstractMessage.QOSType; import io.moquette.server.ConnectionDescriptor; import io.moquette.server.netty.NettyUtils; -import io.moquette.spi.ClientSession; -import io.moquette.spi.IMatchingCondition; -import io.moquette.spi.IMessagesStore; +import io.moquette.spi.*; import io.moquette.spi.IMessagesStore.StoredMessage; -import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -378,7 +375,7 @@ public class ProtocolProcessorTest { m_processor.processConnect(m_channel, connectMessage); //Verify no messages are still stored - Collection guids = m_sessionStore.enqueued(FAKE_PUBLISHER_ID); + Collection guids = m_sessionStore.enqueued(FAKE_PUBLISHER_ID); assertTrue(m_messagesStore.listMessagesInSession(guids).isEmpty()); } diff --git a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java index e2654a69..a38a6ec6 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -24,6 +24,7 @@ import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.ISessionsStore.ClientTopicCouple; import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.subscriptions.Subscription; import org.junit.After; import org.junit.Before; @@ -108,7 +109,7 @@ public class MapDBPersistentStoreTest { //request a first ID int packetId = m_sessionsStore.nextPacketID("CLIENT"); - m_sessionsStore.inFlight("CLIENT", packetId, "ABCDE"); //simulate an inflight + m_sessionsStore.inFlight("CLIENT", packetId, new MessageGUID("ABCDE")); //simulate an inflight assertEquals(1, packetId); //release the ID @@ -137,7 +138,7 @@ public class MapDBPersistentStoreTest { publishToStore.setClientID(TEST_CLIENT); publishToStore.setMessageID(1); publishToStore.setRetained(false); - String guid = m_messagesStore.storePublishForFuture(publishToStore); + MessageGUID guid = m_messagesStore.storePublishForFuture(publishToStore); //Exercise m_messagesStore.dropMessagesInSession("TestClient"); @@ -155,7 +156,7 @@ public class MapDBPersistentStoreTest { publishToStore.setClientID(TEST_CLIENT); publishToStore.setMessageID(1); publishToStore.setRetained(true); - String guid = m_messagesStore.storePublishForFuture(publishToStore); + MessageGUID guid = m_messagesStore.storePublishForFuture(publishToStore); //Exercise m_messagesStore.dropMessagesInSession("TestClient"); diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index 937fd310..c1314f66 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -19,6 +19,7 @@ import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.Utils; import io.moquette.spi.impl.subscriptions.Subscription; @@ -39,17 +40,17 @@ public class MemorySessionStore implements ISessionsStore { private Map m_persistentSessions = new HashMap<>(); //maps clientID->[MessageId -> guid] - private Map> m_inflightStore = new HashMap<>(); + private Map> m_inflightStore = new HashMap<>(); // private Map> m_inflightIDs = new HashMap<>(); //maps clientID->[guid*] - private Map> m_enqueuedStore = new HashMap<>(); + private Map> m_enqueuedStore = new HashMap<>(); //maps clientID->[MessageId -> guid] - private Map> m_secondPhaseStore = new HashMap<>(); + private Map> m_secondPhaseStore = new HashMap<>(); - private Map> m_messageToGuids; + private Map> m_messageToGuids; private final IMessagesStore m_messagesStore; - public MemorySessionStore(IMessagesStore messagesStore, Map> messageToGuids) { + public MemorySessionStore(IMessagesStore messagesStore, Map> messageToGuids) { m_messageToGuids = messageToGuids; this.m_messagesStore = messagesStore; } @@ -158,7 +159,7 @@ public class MemorySessionStore implements ISessionsStore { @Override public void inFlightAck(String clientID, int messageID) { - Map m = this.m_inflightStore.get(clientID); + Map m = this.m_inflightStore.get(clientID); if (m == null) { LOG.error("Can't find the inFlight record for client <{}>", clientID); return; @@ -167,8 +168,8 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public void inFlight(String clientID, int messageID, String guid) { - Map m = this.m_inflightStore.get(clientID); + public void inFlight(String clientID, int messageID, MessageGUID guid) { + Map m = this.m_inflightStore.get(clientID); if (m == null) { m = new HashMap<>(); } @@ -181,7 +182,7 @@ public class MemorySessionStore implements ISessionsStore { * */ @Override public int nextPacketID(String clientID) { - Map m = this.m_inflightStore.get(clientID); + Map m = this.m_inflightStore.get(clientID); if (m == null) { m = new HashMap<>(); int nextPacketId = 1; @@ -195,20 +196,20 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public void bindToDeliver(String guid, String clientID) { - Set guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); + public void bindToDeliver(MessageGUID guid, String clientID) { + Set guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); guids.add(guid); m_enqueuedStore.put(clientID, guids); } @Override - public Collection enqueued(String clientID) { - return Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); + public Collection enqueued(String clientID) { + return Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); } @Override - public void removeEnqueued(String clientID, String guid) { - Set guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); + public void removeEnqueued(String clientID, MessageGUID guid) { + Set guids = Utils.defaultGet(m_enqueuedStore, clientID, new HashSet()); guids.remove(guid); m_enqueuedStore.put(clientID, guids); } @@ -216,31 +217,31 @@ public class MemorySessionStore implements ISessionsStore { @Override public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); - Map m = this.m_inflightStore.get(clientID); + Map m = this.m_inflightStore.get(clientID); if (m == null) { LOG.error("Can't find the inFlight record for client <{}>", clientID); return; } - String guid = m.remove(messageID); + MessageGUID guid = m.remove(messageID); LOG.info("Moving to second phase store"); - Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); messageIDs.put(messageID, guid); m_secondPhaseStore.put(clientID, messageIDs); } @Override - public String secondPhaseAcknowledged(String clientID, int messageID) { - Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); - String guid = messageIDs.remove(messageID); + public MessageGUID secondPhaseAcknowledged(String clientID, int messageID) { + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); + MessageGUID guid = messageIDs.remove(messageID); m_secondPhaseStore.put(clientID, messageIDs); return guid; } @Override - public String mapToGuid(String clientID, int messageID) { - HashMap guids = (HashMap) Utils.defaultGet(m_messageToGuids, - clientID, new HashMap()); + public MessageGUID mapToGuid(String clientID, int messageID) { + HashMap guids = (HashMap) Utils.defaultGet(m_messageToGuids, + clientID, new HashMap()); return guids.get(messageID); }