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 5d554868..30db1473 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -141,7 +141,7 @@ public interface IMessagesStore { * */ int nextPacketID(String clientID); - void close(); +// void close(); void cleanRetained(String topic); } 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 1f953d01..b7cc6bdb 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java @@ -28,6 +28,8 @@ import org.eclipse.moquette.spi.impl.subscriptions.Subscription; */ public interface ISessionsStore { + void initStore(); + /** * Add a new subscription to the 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 c79533c2..703e399a 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 @@ -626,6 +626,7 @@ public class ProtocolProcessor { subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription + //TODO this is ugly, it does a linear scan on potential big dataset Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { public boolean match(String key) { return SubscriptionsStore.matchTopics(key, topic); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java index 8cc15ca7..ced6859b 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java @@ -48,9 +48,7 @@ public class SimpleMessaging { private SubscriptionsStore subscriptions; - - private IMessagesStore m_storageService; - private ISessionsStore m_sessionsStore; + private MapDBPersistentStore m_mapStorage; private BrokerInterceptor m_interceptor; @@ -72,11 +70,10 @@ public class SimpleMessaging { subscriptions = new SubscriptionsStore(); //TODO use a property to select the storage path - MapDBPersistentStore mapStorage = new MapDBPersistentStore(props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, "")); - m_storageService = mapStorage; - m_sessionsStore = mapStorage; - - m_storageService.initStore(); + m_mapStorage = new MapDBPersistentStore(props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, "")); + m_mapStorage.initStore(); + IMessagesStore messagesStore = m_mapStorage.messagesStore(); + ISessionsStore sessionsStore = m_mapStorage.sessionsStore(messagesStore); List observers = new ArrayList<>(); String interceptorClassName = props.getProperty("intercept.handler"); @@ -90,7 +87,7 @@ public class SimpleMessaging { } m_interceptor = new BrokerInterceptor(observers); - subscriptions.init(m_sessionsStore); + subscriptions.init(sessionsStore); String configPath = System.getProperty("moquette.path", null); String authenticatorClassName = props.getProperty(Constants.AUTHENTICATOR_CLASS_NAME, ""); @@ -136,7 +133,7 @@ public class SimpleMessaging { } boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); - m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor); + m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor); return m_processor; } @@ -176,6 +173,6 @@ public class SimpleMessaging { } public void shutdown() { - this.m_storageService.close(); + this.m_mapStorage.close(); } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBMessagesStore.java new file mode 100644 index 00000000..e8bfd796 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBMessagesStore.java @@ -0,0 +1,138 @@ +/* + * 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.persistence; + +import org.eclipse.moquette.spi.IMatchingCondition; +import org.eclipse.moquette.spi.IMessagesStore; +import org.mapdb.DB; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; +import java.util.concurrent.ConcurrentMap; + +import static org.eclipse.moquette.spi.persistence.MapDBSessionsStore.messageId2GuidsMapName; + +/** + * IMessagesStore implementation backed by MapDB. + * + * @author andrea + */ +class MapDBMessagesStore implements IMessagesStore { + + private static final Logger LOG = LoggerFactory.getLogger(MapDBMessagesStore.class); + + private DB m_db; + + //map clientID <-> set of currently in flight packet identifiers + private Map> m_inFlightIds; + //maps clientID -> guid + private ConcurrentMap m_retainedStore; + //maps guid to message, it's message store + private ConcurrentMap m_persistentMessageStore; + + + MapDBMessagesStore(DB db) { + m_db = db; + } + + @Override + public void initStore() { + m_retainedStore = m_db.getHashMap("retained"); + m_persistentMessageStore = m_db.getHashMap("persistedMessages"); + m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); + } + + @Override + public void storeRetained(String topic, String guid) { + m_retainedStore.put(topic, guid); + } + + @Override + public Collection searchMatching(IMatchingCondition condition) { + 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(); + StoredMessage storedMsg = m_persistentMessageStore.get(guid); + if (condition.match(entry.getKey())) { + results.add(storedMsg); + } + } + + return results; + } + + @Override + 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, 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<>(); + for (String guid : guids) { + ret.add(m_persistentMessageStore.get(guid)); + } + return ret; + } + + @Override + public void dropMessagesInSession(String clientID) { + m_db.getHashMap(messageId2GuidsMapName(clientID)).clear(); + m_persistentMessageStore.remove(clientID); + } + + @Override + public StoredMessage getMessageByGuid(String guid) { + return m_persistentMessageStore.get(guid); + } + + /** + * Return the next valid packetIdentifier for the given client session. + * */ + @Override + public int nextPacketID(String clientID) { + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient == null) { + int nextPacketId = 1; + inFlightForClient = new HashSet<>(); + inFlightForClient.add(nextPacketId); + this.m_inFlightIds.put(clientID, inFlightForClient); + return nextPacketId; + } + int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); + int nextPacketId = (maxId + 1) % 0xFFFF; + inFlightForClient.add(nextPacketId); + return nextPacketId; + } + + @Override + public void cleanRetained(String topic) { + m_retainedStore.remove(topic); + } +} 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 59581ee6..a6b8bf52 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,11 +17,8 @@ package org.eclipse.moquette.spi.persistence; import org.eclipse.moquette.proto.MQTTException; -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.subscriptions.Subscription; import org.mapdb.DB; import org.mapdb.DBMaker; import org.slf4j.Logger; @@ -30,18 +27,30 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.io.Serializable; -import java.util.*; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import static org.eclipse.moquette.spi.impl.Utils.defaultGet; - /** * MapDB main persistence implementation */ -public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { +public class MapDBPersistentStore { + + /** + * Factory method to create message store backed by MapDB + * */ + public IMessagesStore messagesStore() { + //TODO check m_db is valid and + IMessagesStore msgStore = new MapDBMessagesStore(m_db); + msgStore.initStore(); + return msgStore; + } + + public ISessionsStore sessionsStore(IMessagesStore msgStore) { + ISessionsStore sessionsStore = new MapDBSessionsStore(m_db, msgStore); + sessionsStore.initStore(); + return sessionsStore; + } /** * This is a DTO used to persist minimal status (clean session and activation status) of @@ -59,63 +68,30 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class); - //maps clientID -> guid - private ConcurrentMap m_retainedStore; - //maps guid to message, it's message store - private ConcurrentMap m_persistentMessageStore; - //maps clientID->[MessageId -> guid] - private ConcurrentMap> m_inflightStore; - //map clientID <-> set of currently in flight packet identifiers - Map> m_inFlightIds; - //persistent Map of clientID, set of Subscriptions - private ConcurrentMap> m_persistentSubscriptions; - private ConcurrentMap m_persistentSessions; - - //maps clientID->[guid*], insertion order cares, it's queue - private ConcurrentMap> m_enqueuedStore; - //maps clientID->[messageID*] - private ConcurrentMap> m_secondPhaseStore; - private DB m_db; private String m_storePath; protected final ScheduledExecutorService m_scheduler = Executors.newScheduledThreadPool(1); - /* - * The default constructor will create an in memory store as no file path was specified - */ - public MapDBPersistentStore() { - this.m_storePath = ""; - } - public MapDBPersistentStore(String storePath) { this.m_storePath = storePath; } - @Override public void initStore() { - if (m_storePath == null || m_storePath.isEmpty()) { - m_db = DBMaker.newMemoryDB().make(); - } else { - File tmpFile; - try { - tmpFile = new File(m_storePath); - boolean fileNewlyCreated = tmpFile.createNewFile(); + if (m_storePath == null || m_storePath.isEmpty()) { + m_db = DBMaker.newMemoryDB().make(); + } else { + File tmpFile; + try { + tmpFile = new File(m_storePath); + boolean fileNewlyCreated = tmpFile.createNewFile(); LOG.info("Starting with {} [{}] db file", fileNewlyCreated ? "fresh" : "existing", m_storePath); - } catch (IOException ex) { - LOG.error(null, ex); - throw new MQTTException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex); - } - m_db = DBMaker.newFileDB(tmpFile).make(); - } - m_retainedStore = m_db.getHashMap("retained"); - m_persistentMessageStore = m_db.getHashMap("persistedMessages"); - m_inflightStore = m_db.getHashMap("inflight"); - m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); - m_persistentSubscriptions = m_db.getHashMap("subscriptions"); - m_persistentSessions = m_db.getHashMap("sessions"); - m_enqueuedStore = m_db.getHashMap("sessionQueue"); - m_secondPhaseStore = m_db.getHashMap("secondPhase"); + } catch (IOException ex) { + LOG.error(null, ex); + throw new MQTTException("Can't create temp file for subscriptions storage [" + m_storePath + "]", ex); + } + m_db = DBMaker.newFileDB(tmpFile).make(); + } m_scheduler.scheduleWithFixedDelay(new Runnable() { @Override public void run() { @@ -124,287 +100,13 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { }, 30, 30, TimeUnit.SECONDS); } - @Override - public void cleanRetained(String topic) { - m_retainedStore.remove(topic); - } - - @Override - public void storeRetained(String topic, String guid) { - m_retainedStore.put(topic, guid); - } - - @Override - public Collection searchMatching(IMatchingCondition condition) { - 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(); - StoredMessage storedMsg = m_persistentMessageStore.get(guid); - if (condition.match(entry.getKey())) { - results.add(storedMsg); - } - } - - return results; - } - - @Override - 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, 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<>(); - for (String guid : guids) { - ret.add(m_persistentMessageStore.get(guid)); - } - return ret; - } - - @Override - public void dropMessagesInSession(String clientID) { - m_db.getHashMap(messageId2GuidsMapName(clientID)).clear(); - m_persistentMessageStore.remove(clientID); - } - - @Override - public StoredMessage getMessageByGuid(String guid) { - return m_persistentMessageStore.get(guid); - } - - @Override - public String mapToGuid(String clientID, int messageID) { - 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. - * */ - @Override - public int nextPacketID(String clientID) { - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient == null) { - int nextPacketId = 1; - inFlightForClient = new HashSet<>(); - inFlightForClient.add(nextPacketId); - this.m_inFlightIds.put(clientID, inFlightForClient); - return nextPacketId; - } - int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); - int nextPacketId = (maxId + 1) % 0xFFFF; - inFlightForClient.add(nextPacketId); - return nextPacketId; - } - - @Override - public void removeSubscription(String topic, String clientID) { - LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); - if (!m_persistentSubscriptions.containsKey(clientID)) { - return; - } - Set clientSubscriptions = m_persistentSubscriptions.get(clientID); - //search for the subscription to remove - Subscription toBeRemoved = null; - for (Subscription sub : clientSubscriptions) { - if (sub.getTopicFilter().equals(topic)) { - toBeRemoved = sub; - break; - } - } - - if (toBeRemoved != null) { - clientSubscriptions.remove(toBeRemoved); - } - m_persistentSubscriptions.put(clientID, clientSubscriptions); - if (LOG.isTraceEnabled()) { - LOG.trace("persisted subscriptions {}", m_persistentSubscriptions); - } - } - - @Override - public void addNewSubscription(Subscription newSubscription) { - LOG.debug("addNewSubscription invoked with subscription {}", newSubscription); - final String clientID = newSubscription.getClientId(); - if (!m_persistentSubscriptions.containsKey(clientID)) { - LOG.debug("subscriptions for clientID <{}> not present, empty subscriptions set", clientID); - m_persistentSubscriptions.put(clientID, new HashSet()); - } - - Set subs = m_persistentSubscriptions.get(clientID); - if (!subs.contains(newSubscription)) { - LOG.debug("updating <{}> subscriptions set with new subscription", clientID); - //TODO check the subs doesn't contain another subscription to the same topic with different - Subscription existingSubscription = null; - for (Subscription scanSub : subs) { - if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) { - existingSubscription = scanSub; - break; - } - } - if (existingSubscription != null) { - subs.remove(existingSubscription); - } - subs.add(newSubscription); - m_persistentSubscriptions.put(clientID, subs); - LOG.debug("clientID {} subscriptions set now is {}", clientID, subs); - } - } - - @Override - public void wipeSubscriptions(String clientID) { - m_persistentSubscriptions.remove(clientID); - } - - @Override - public void updateSubscriptions(String clientID, Set subscriptions) { - m_persistentSubscriptions.put(clientID, subscriptions); - } - - public List listAllSubscriptions() { - List allSubscriptions = new ArrayList<>(); - for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { - allSubscriptions.addAll(entry.getValue()); - } - LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions); - return allSubscriptions; - } - - @Override - public boolean contains(String clientID) { - return m_persistentSubscriptions.containsKey(clientID); - } - - @Override - public ClientSession createNewSession(String clientID, boolean cleanSession) { - LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession); - if (m_persistentSessions.containsKey(clientID)) { - LOG.error("already exists a session for client <{}>, bad condition", clientID); - throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); - } - LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); - m_persistentSubscriptions.put(clientID, new HashSet()); - m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false)); - return new ClientSession(clientID, this, this, cleanSession); - } - - @Override - public ClientSession sessionForClient(String clientID) { - if (!m_persistentSessions.containsKey(clientID)) { - return null; - } - - PersistentSession storedSession = m_persistentSessions.get(clientID); - ClientSession clientSession = new ClientSession(clientID, this, this, storedSession.cleanSession); - if (storedSession.active) { - clientSession.activate(); - } - return clientSession; - } - - @Override - public void activate(String clientID) { - activationHelper(clientID, true); - } - - @Override - public void deactivate(String clientID) { - activationHelper(clientID, false); - } - - @Override - public void inFlightAck(String clientID, int messageID) { - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - LOG.error("Can't find the inFlight record for client <{}>", clientID); - return; - } - m.remove(messageID); - - //remove from the ids store - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient != null) { - inFlightForClient.remove(messageID); - } - } - - @Override - public void inFlight(String clientID, int messageID, String guid) { - Map m = this.m_inflightStore.get(clientID); - if (m == null) { - m = new HashMap<>(); - } - m.put(messageID, guid); - this.m_inflightStore.put(clientID, m); - } - - @Override - public void bindToDeliver(String guid, String clientID) { - List guids = defaultGet(m_enqueuedStore, clientID, new ArrayList()); - guids.add(guid); - m_enqueuedStore.put(clientID, guids); - } - - @Override - public Collection enqueued(String clientID) { - return defaultGet(m_enqueuedStore, clientID, new ArrayList()); - } - - @Override - public void removeEnqueued(String clientID, String guid) { - List guids = defaultGet(m_enqueuedStore, clientID, new ArrayList()); - guids.remove(guid); - m_enqueuedStore.put(clientID, guids); - } - - @Override - public void secondPhaseAcknowledged(String clientID, int messageID) { - Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.remove(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - } - - @Override - public void secondPhaseAckWaiting(String clientID, int messageID) { - Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.add(messageID); - m_secondPhaseStore.put(clientID, messageIDs); - } - - private void activationHelper(String clientID, boolean activation) { - PersistentSession storedSession = m_persistentSessions.get(clientID); - if (storedSession == null) { - throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null); - } - PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation); - m_persistentSessions.put(clientID, newStoredSession); - } - - @Override public void close() { if (this.m_db.isClosed()) { LOG.debug("already closed"); return; } this.m_db.commit(); - LOG.debug("persisted subscriptions {}", m_persistentSubscriptions); + //LOG.debug("persisted subscriptions {}", m_persistentSubscriptions); this.m_db.close(); LOG.debug("closed disk storage"); this.m_scheduler.shutdown(); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBSessionsStore.java new file mode 100644 index 00000000..d10b34a6 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBSessionsStore.java @@ -0,0 +1,264 @@ +/* + * 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.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.subscriptions.Subscription; +import org.eclipse.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; +import org.mapdb.DB; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; +import java.util.concurrent.ConcurrentMap; + +import static org.eclipse.moquette.spi.impl.Utils.defaultGet; + +/** + * ISessionsStore implementation backed by MapDB. + * + * @author andrea + */ +class MapDBSessionsStore implements ISessionsStore { + + private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class); + + //maps clientID->[MessageId -> guid] + private ConcurrentMap> m_inflightStore; + //map clientID <-> set of currently in flight packet identifiers + private Map> m_inFlightIds; + //persistent Map of clientID, set of Subscriptions + private ConcurrentMap> m_persistentSubscriptions; + private ConcurrentMap m_persistentSessions; + //maps clientID->[guid*], insertion order cares, it's queue + private ConcurrentMap> m_enqueuedStore; + //maps clientID->[messageID*] + private ConcurrentMap> m_secondPhaseStore; + + private final DB m_db; + private final IMessagesStore m_messagesStore; + + MapDBSessionsStore(DB db, IMessagesStore messagesStore) { + m_db = db; + m_messagesStore = messagesStore; + } + + @Override + public void initStore() { + m_inflightStore = m_db.getHashMap("inflight"); + m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); + m_persistentSubscriptions = m_db.getHashMap("subscriptions"); + m_persistentSessions = m_db.getHashMap("sessions"); + m_enqueuedStore = m_db.getHashMap("sessionQueue"); + m_secondPhaseStore = m_db.getHashMap("secondPhase"); + } + + @Override + public void addNewSubscription(Subscription newSubscription) { + LOG.debug("addNewSubscription invoked with subscription {}", newSubscription); + final String clientID = newSubscription.getClientId(); + if (!m_persistentSubscriptions.containsKey(clientID)) { + LOG.debug("subscriptions for clientID <{}> not present, empty subscriptions set", clientID); + m_persistentSubscriptions.put(clientID, new HashSet()); + } + + Set subs = m_persistentSubscriptions.get(clientID); + if (!subs.contains(newSubscription)) { + LOG.debug("updating <{}> subscriptions set with new subscription", clientID); + //TODO check the subs doesn't contain another subscription to the same topic with different + Subscription existingSubscription = null; + for (Subscription scanSub : subs) { + if (newSubscription.getTopicFilter().equals(scanSub.getTopicFilter())) { + existingSubscription = scanSub; + break; + } + } + if (existingSubscription != null) { + subs.remove(existingSubscription); + } + subs.add(newSubscription); + m_persistentSubscriptions.put(clientID, subs); + LOG.debug("clientID {} subscriptions set now is {}", clientID, subs); + } + } + + @Override + public void removeSubscription(String topic, String clientID) { + LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); + if (!m_persistentSubscriptions.containsKey(clientID)) { + return; + } + Set clientSubscriptions = m_persistentSubscriptions.get(clientID); + //search for the subscription to remove + Subscription toBeRemoved = null; + for (Subscription sub : clientSubscriptions) { + if (sub.getTopicFilter().equals(topic)) { + toBeRemoved = sub; + break; + } + } + + if (toBeRemoved != null) { + clientSubscriptions.remove(toBeRemoved); + } + m_persistentSubscriptions.put(clientID, clientSubscriptions); + if (LOG.isTraceEnabled()) { + LOG.trace("persisted subscriptions {}", m_persistentSubscriptions); + } + } + + @Override + public void wipeSubscriptions(String clientID) { + m_persistentSubscriptions.remove(clientID); + } + + @Override + public void updateSubscriptions(String clientID, Set subscriptions) { + m_persistentSubscriptions.put(clientID, subscriptions); + } + + @Override + public List listAllSubscriptions() { + List allSubscriptions = new ArrayList<>(); + for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { + allSubscriptions.addAll(entry.getValue()); + } + LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions); + return allSubscriptions; + } + + @Override + public boolean contains(String clientID) { + return m_persistentSubscriptions.containsKey(clientID); + } + + @Override + public ClientSession createNewSession(String clientID, boolean cleanSession) { + LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientID, cleanSession); + if (m_persistentSessions.containsKey(clientID)) { + LOG.error("already exists a session for client <{}>, bad condition", clientID); + throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientID); + } + LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID); + m_persistentSubscriptions.put(clientID, new HashSet()); + m_persistentSessions.putIfAbsent(clientID, new PersistentSession(cleanSession, false)); + return new ClientSession(clientID, m_messagesStore, this, cleanSession); + } + + @Override + public ClientSession sessionForClient(String clientID) { + if (!m_persistentSessions.containsKey(clientID)) { + return null; + } + + PersistentSession storedSession = m_persistentSessions.get(clientID); + ClientSession clientSession = new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession); + if (storedSession.active) { + clientSession.activate(); + } + return clientSession; + } + + @Override + public void activate(String clientID) { + activationHelper(clientID, true); + } + + @Override + public void deactivate(String clientID) { + activationHelper(clientID, false); + } + + private void activationHelper(String clientID, boolean activation) { + PersistentSession storedSession = m_persistentSessions.get(clientID); + if (storedSession == null) { + throw new IllegalStateException((activation ? "activating" : "deactivating") + " a session never stored/created, clientID <"+ clientID + ">", null); + } + PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation); + m_persistentSessions.put(clientID, newStoredSession); + } + + @Override + public void inFlightAck(String clientID, int messageID) { + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + LOG.error("Can't find the inFlight record for client <{}>", clientID); + return; + } + m.remove(messageID); + + //remove from the ids store + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient != null) { + inFlightForClient.remove(messageID); + } + } + + @Override + public void inFlight(String clientID, int messageID, String guid) { + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + m = new HashMap<>(); + } + m.put(messageID, guid); + this.m_inflightStore.put(clientID, m); + } + + @Override + public void bindToDeliver(String guid, String clientID) { + List guids = defaultGet(m_enqueuedStore, clientID, new ArrayList()); + guids.add(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public Collection enqueued(String clientID) { + return defaultGet(m_enqueuedStore, clientID, new ArrayList()); + } + + @Override + public void removeEnqueued(String clientID, String guid) { + List guids = defaultGet(m_enqueuedStore, clientID, new ArrayList()); + guids.remove(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public void secondPhaseAcknowledged(String clientID, int messageID) { + Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.remove(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public void secondPhaseAckWaiting(String clientID, int messageID) { + Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.add(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public String mapToGuid(String clientID, int messageID) { + ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); + return messageIdToGuid.get(messageID); + } + + static String messageId2GuidsMapName(String clientID) { + return "guidsMapping_" + clientID; + } +} 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 0a38e961..13c7e77d 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 @@ -136,11 +136,6 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { return nextPacketId; } - @Override - public void close() { - //To change body of implemented methods use File | Settings | File Templates. - } - @Override public void addNewSubscription(Subscription newSubscription) { m_sessionsStore.addNewSubscription(newSubscription); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java index 1cfd619f..164892de 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -17,6 +17,8 @@ package org.eclipse.moquette.spi.persistence; import org.eclipse.moquette.proto.messages.AbstractMessage; import org.eclipse.moquette.server.IntegrationUtils; +import org.eclipse.moquette.spi.IMessagesStore; +import org.eclipse.moquette.spi.ISessionsStore; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.junit.After; import org.junit.Before; @@ -35,12 +37,16 @@ import static org.junit.Assert.*; public class MapDBPersistentStoreTest { MapDBPersistentStore m_storageService; + ISessionsStore m_sessionsStore; + IMessagesStore m_messagesStore; @Before public void setUp() throws Exception { IntegrationUtils.cleanPersistenceFile(DEFAULT_PERSISTENT_PATH); m_storageService = new MapDBPersistentStore(DEFAULT_PERSISTENT_PATH); m_storageService.initStore(); + m_messagesStore = m_storageService.messagesStore(); + m_sessionsStore = m_storageService.sessionsStore(m_messagesStore); } @After @@ -55,12 +61,12 @@ public class MapDBPersistentStoreTest { @Test public void overridingSubscriptions() { Subscription oldSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); - m_storageService.addNewSubscription(oldSubscription); + m_sessionsStore.addNewSubscription(oldSubscription); Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE, false); - m_storageService.addNewSubscription(overrindingSubscription); + m_sessionsStore.addNewSubscription(overrindingSubscription); //Verify - List subscriptions = m_storageService.listAllSubscriptions(); + List subscriptions = m_sessionsStore.listAllSubscriptions(); assertEquals(1, subscriptions.size()); Subscription sub = subscriptions.get(0); assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos()); @@ -68,18 +74,18 @@ public class MapDBPersistentStoreTest { @Test public void testNextPacketID_notExistingClientSession() { - int packetId = m_storageService.nextPacketID("NOT_EXISTING_CLI"); + int packetId = m_messagesStore.nextPacketID("NOT_EXISTING_CLI"); assertEquals(1, packetId); } @Test public void testNextPacketID_existingClientSession() { //Force creation of inflight map for the CLIENT session - int packetId = m_storageService.nextPacketID("CLIENT"); + int packetId = m_messagesStore.nextPacketID("CLIENT"); assertEquals(1, packetId); //request a second packetID - packetId = m_storageService.nextPacketID("CLIENT"); + packetId = m_messagesStore.nextPacketID("CLIENT"); assertEquals(2, packetId); } @@ -87,15 +93,15 @@ public class MapDBPersistentStoreTest { public void testNextPacketID() { //request a first ID - int packetId = m_storageService.nextPacketID("CLIENT"); - m_storageService.inFlight("CLIENT", packetId, "ABCDE"); //simulate an inflight + int packetId = m_messagesStore.nextPacketID("CLIENT"); + m_sessionsStore.inFlight("CLIENT", packetId, "ABCDE"); //simulate an inflight assertEquals(1, packetId); //release the ID - m_storageService.inFlightAck("CLIENT", packetId); + m_sessionsStore.inFlightAck("CLIENT", packetId); //request a second packetID, counter restarts from 0 - packetId = m_storageService.nextPacketID("CLIENT"); + packetId = m_messagesStore.nextPacketID("CLIENT"); assertEquals(1, packetId); } 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 97a40f05..99109d36 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 @@ -73,6 +73,11 @@ public class MemorySessionStore implements ISessionsStore { } } + @Override + public void initStore() { + + } + @Override public void addNewSubscription(Subscription newSubscription) { final String clientID = newSubscription.getClientId();