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 a6b8bf52..36e7c036 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 @@ -36,22 +36,6 @@ import java.util.concurrent.TimeUnit; */ 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 * a session. @@ -76,6 +60,22 @@ public class MapDBPersistentStore { public MapDBPersistentStore(String storePath) { this.m_storePath = storePath; } + + /** + * 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; + } public void initStore() { if (m_storePath == null || m_storePath.isEmpty()) { diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryMessagesStore.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryMessagesStore.java new file mode 100644 index 00000000..b8a03ef1 --- /dev/null +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryMessagesStore.java @@ -0,0 +1,123 @@ +/* + * 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; + +import org.eclipse.moquette.spi.IMatchingCondition; +import org.eclipse.moquette.spi.IMessagesStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; + +import static org.eclipse.moquette.spi.impl.Utils.defaultGet; + +/** + * @author andrea + */ +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_inflightIDs = new HashMap<>(); + + MemoryMessagesStore(Map> messageToGuids) { + m_messageToGuids = messageToGuids; + } + + @Override + public void initStore() { + } + + @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); + String guid = UUID.randomUUID().toString(); + evt.setGuid(guid); + m_persistentMessageStore.put(guid, evt); + HashMap guids = (HashMap) defaultGet(m_messageToGuids, + evt.getClientID(), new HashMap()); + guids.put(evt.getMessageID(), guid); + return guid; + } + + @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_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 = m_inflightIDs.get(clientID); + if (inFlightForClient == null) { + int nextPacketId = 1; + inFlightForClient = new HashSet<>(); + inFlightForClient.add(nextPacketId); + m_inflightIDs.put(clientID, inFlightForClient); + return nextPacketId; + } + int maxId = 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/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index 13c7e77d..22407e6a 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,227 +15,29 @@ */ package org.eclipse.moquette.spi.impl; -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.persistence.MemorySessionStore; -import org.eclipse.moquette.spi.impl.subscriptions.Subscription; - import org.eclipse.moquette.spi.ISessionsStore; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.eclipse.moquette.spi.persistence.MemorySessionStore; +import java.util.HashMap; +import java.util.Map; -import static org.eclipse.moquette.spi.impl.Utils.defaultGet; - -/** - */ -public class MemoryStorageService implements IMessagesStore, ISessionsStore { +public class MemoryStorageService { - 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<>(); - //maps clientID->[MessageId -> guid] - private Map> m_inflightStore = new HashMap<>(); - //maps clientID->[guid*] - private Map> m_enqueuedStore = new HashMap<>(); - //maps clientID->[messageID*] - private Map> m_secondPhaseStore = new HashMap<>(); - private Map> m_inflightIDs = new HashMap<>(); - private Map> m_messageToGuids = new HashMap<>(); private MemorySessionStore m_sessionsStore; + private MemoryMessagesStore m_messagesStore; - private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class); - public void initStore() { - m_sessionsStore = new MemorySessionStore(this); - } - - @Override - public void cleanRetained(String topic) { - m_retainedStore.remove(topic); - } - - @Override - public void storeRetained(String topic, String guid) { - m_retainedStore.put(topic, guid); + Map> messageToGuids = new HashMap<>(); + m_messagesStore = new MemoryMessagesStore(messageToGuids); + m_sessionsStore = new MemorySessionStore(m_messagesStore, messageToGuids); } - @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; + public IMessagesStore messagesStore() { + return m_messagesStore; } - @Override - public String storePublishForFuture(StoredMessage evt) { - LOG.debug("storePublishForFuture store evt {}", evt); - String guid = UUID.randomUUID().toString(); - evt.setGuid(guid); - m_persistentMessageStore.put(guid, evt); - HashMap guids = (HashMap) defaultGet(m_messageToGuids, - evt.getClientID(), new HashMap()); - guids.put(evt.getMessageID(), guid); - return guid; + public ISessionsStore sessionsStore() { + return m_sessionsStore; } - @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_persistentMessageStore.remove(clientID); - } - - @Override - public StoredMessage getMessageByGuid(String guid) { - return m_persistentMessageStore.get(guid); - } - - @Override - public String mapToGuid(String clientID, int messageID) { - HashMap guids = (HashMap) defaultGet(m_messageToGuids, - clientID, new HashMap()); - return guids.get(messageID); - } - - /** - * Return the next valid packetIdentifier for the given client session. - * */ - @Override - public int nextPacketID(String clientID) { - Set inFlightForClient = m_inflightIDs.get(clientID); - if (inFlightForClient == null) { - int nextPacketId = 1; - inFlightForClient = new HashSet<>(); - inFlightForClient.add(nextPacketId); - m_inflightIDs.put(clientID, inFlightForClient); - return nextPacketId; - } - int maxId = Collections.max(inFlightForClient); - int nextPacketId = (maxId + 1) % 0xFFFF; - inFlightForClient.add(nextPacketId); - return nextPacketId; - } - - @Override - public void addNewSubscription(Subscription newSubscription) { - m_sessionsStore.addNewSubscription(newSubscription); - } - - @Override - public void removeSubscription(String topic, String clientID) { - m_sessionsStore.removeSubscription(topic, clientID); - } - - @Override - public void wipeSubscriptions(String sessionID) { - m_sessionsStore.wipeSubscriptions(sessionID); - } - - @Override - public void updateSubscriptions(String clientID, Set subscriptions) { - m_sessionsStore.updateSubscriptions(clientID, subscriptions); - } - - @Override - public List listAllSubscriptions() { - return m_sessionsStore.listAllSubscriptions(); - } - - @Override - public boolean contains(String clientID) { - return m_sessionsStore.contains(clientID); - } - - @Override - public ClientSession createNewSession(String clientID, boolean cleanSession) { - return m_sessionsStore.createNewSession(clientID, cleanSession); - } - - @Override - public ClientSession sessionForClient(String clientID) { - return m_sessionsStore.sessionForClient(clientID); - } - - @Override - public void activate(String clientID) { - m_sessionsStore.activate(clientID); - } - - @Override - public void deactivate(String clientID) { - m_sessionsStore.deactivate(clientID); - } - - @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); - } - - @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) { - Set guids = defaultGet(m_enqueuedStore, clientID, new HashSet()); - guids.add(guid); - m_enqueuedStore.put(clientID, guids); - } - - @Override - public Collection enqueued(String clientID) { - return defaultGet(m_enqueuedStore, clientID, new HashSet()); - } - - @Override - public void removeEnqueued(String clientID, String guid) { - Set guids = defaultGet(m_enqueuedStore, clientID, new HashSet()); - 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); - } } 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 5a4c5f34..57077553 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 @@ -15,29 +15,27 @@ */ package org.eclipse.moquette.spi.impl; +import org.eclipse.moquette.interception.InterceptHandler; +import org.eclipse.moquette.proto.messages.*; +import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType; +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.StoredMessage; +import org.eclipse.moquette.spi.ISessionsStore; +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.junit.Before; +import org.junit.Test; + import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.eclipse.moquette.interception.InterceptHandler; -import org.eclipse.moquette.proto.messages.*; -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.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; import static org.mockito.Mockito.*; /** @@ -63,7 +61,7 @@ public class ProtocolProcessorTest { ConnectMessage connMsg; ProtocolProcessor m_processor; - IMessagesStore m_storageService; + IMessagesStore m_messagesStore; ISessionsStore m_sessionStore; SubscriptionsStore subscriptions; MockAuthenticator m_mockAuthenticator; @@ -79,18 +77,18 @@ public class ProtocolProcessorTest { Thread.sleep(300); MemoryStorageService memStorage = new MemoryStorageService(); memStorage.initStore(); - m_storageService = memStorage; - m_sessionStore = memStorage; - //m_storageService.initStore(); + m_messagesStore = memStorage.messagesStore(); + m_sessionStore = memStorage.sessionsStore(); + //m_messagesStore.initStore(); Map users = new HashMap<>(); users.put(TEST_USER, TEST_PWD); m_mockAuthenticator = new MockAuthenticator(users); subscriptions = new SubscriptionsStore(); - subscriptions.init(memStorage); + subscriptions.init(memStorage.sessionsStore()); m_processor = new ProtocolProcessor(); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); } @@ -115,8 +113,8 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession MemoryStorageService storageService = new MemoryStorageService(); storageService.initStore(); - subs.init(storageService); - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + subs.init(storageService.sessionsStore()); + m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProtocolVersion((byte) 3); connectMessage.setClientID(FAKE_CLIENT_ID); @@ -162,8 +160,8 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession MemoryStorageService storageService = new MemoryStorageService(); storageService.initStore(); - subs.init(storageService); - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + subs.init(storageService.sessionsStore()); + m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); ConnectMessage connectMessage = new ConnectMessage(); @@ -320,10 +318,10 @@ public class ProtocolProcessorTest { }; MemoryStorageService storageService = new MemoryStorageService(); storageService.initStore(); - subs.init(storageService); + subs.init(storageService.sessionsStore()); //simulate a connect that register a clientID to an IoSession - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProtocolVersion((byte) 3); @@ -363,9 +361,9 @@ public class ProtocolProcessorTest { retainedMessage.setRetained(true); retainedMessage.setMessageID(120); retainedMessage.setClientID(FAKE_PUBLISHER_ID); - m_storageService.storePublishForFuture(retainedMessage); + m_messagesStore.storePublishForFuture(retainedMessage); - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProtocolVersion((byte) 3); @@ -374,7 +372,7 @@ public class ProtocolProcessorTest { //Verify no messages are still stored Collection guids = m_sessionStore.enqueued(FAKE_PUBLISHER_ID); - assertTrue(m_storageService.listMessagesInSession(guids).isEmpty()); + assertTrue(m_messagesStore.listMessagesInSession(guids).isEmpty()); } @Test @@ -387,7 +385,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); @@ -413,7 +411,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), + m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise @@ -460,7 +458,7 @@ public class ProtocolProcessorTest { m_processor.processPublish(m_session, cleanPubMsg); //Verify - Collection messages = m_storageService.searchMatching(new IMatchingCondition() { + Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { public boolean match(String key) { return SubscriptionsStore.matchTopics(key, FAKE_TOPIC); } @@ -476,8 +474,10 @@ public class ProtocolProcessorTest { forwardPublish.setRetained(true); forwardPublish.setMessageID(1); - IMessagesStore memoryMessageStore = new MemoryStorageService(); - ISessionsStore sessionsStore = new MemorySessionStore(memoryMessageStore); + MemoryStorageService memStore = new MemoryStorageService(); + memStore.initStore(); + IMessagesStore memoryMessageStore = memStore.messagesStore(); + ISessionsStore sessionsStore = memStore.sessionsStore(); sessionsStore.createNewSession("Sub A", false).activate(); sessionsStore.createNewSession("Sub B", false).activate(); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 39c323ea..42b44177 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -44,7 +44,7 @@ public class ProtocolProcessor_CONNECT_Test { ConnectMessage connMsg; ProtocolProcessor m_processor; - IMessagesStore m_storageService; + IMessagesStore m_messagesStore; ISessionsStore m_sessionStore; SubscriptionsStore subscriptions; MockAuthenticator m_mockAuthenticator; @@ -60,18 +60,18 @@ public class ProtocolProcessor_CONNECT_Test { Thread.sleep(300); MemoryStorageService memStorage = new MemoryStorageService(); memStorage.initStore(); - m_storageService = memStorage; - m_sessionStore = memStorage; - //m_storageService.initStore(); + m_messagesStore = memStorage.messagesStore(); + m_sessionStore = memStorage.sessionsStore(); + //m_messagesStore.initStore(); Map users = new HashMap<>(); users.put(TEST_USER, TEST_PWD); m_mockAuthenticator = new MockAuthenticator(users); subscriptions = new SubscriptionsStore(); - subscriptions.init(memStorage); + subscriptions.init(m_sessionStore); m_processor = new ProtocolProcessor(); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); } @@ -163,7 +163,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void prohibitAnonymousClient() { connMsg.setClientID("123"); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); @@ -177,7 +177,7 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setClientID("123"); connMsg.setUserFlag(true); connMsg.setUsername(TEST_USER + "_fake"); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); @@ -189,7 +189,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void acceptAnonymousClient() { connMsg.setClientID("123"); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index 07523534..08b6d972 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -20,6 +20,7 @@ import java.text.ParseException; import java.util.Arrays; import java.util.List; +import org.eclipse.moquette.spi.ISessionsStore; import org.eclipse.moquette.spi.impl.MemoryStorageService; import org.eclipse.moquette.proto.messages.AbstractMessage; @@ -43,7 +44,7 @@ public class SubscriptionsStoreTest { store = new SubscriptionsStore(); MemoryStorageService storageService = new MemoryStorageService(); storageService.initStore(); - store.init(storageService); + store.init(storageService.sessionsStore()); } @Test @@ -235,7 +236,7 @@ public class SubscriptionsStoreTest { store = new SubscriptionsStore(); MemoryStorageService memStore = new MemoryStorageService(); memStore.initStore(); - store.init(memStore); + store.init(memStore.sessionsStore()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); store.add(sub); assertFalse(store.matches(topic).isEmpty()); @@ -245,7 +246,7 @@ public class SubscriptionsStoreTest { store = new SubscriptionsStore(); MemoryStorageService memStore = new MemoryStorageService(); memStore.initStore(); - store.init(memStore); + store.init(memStore.sessionsStore()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); store.add(sub); assertTrue(store.matches(topic).isEmpty()); @@ -346,11 +347,12 @@ public class SubscriptionsStoreTest { SubscriptionsStore aStore = new SubscriptionsStore(); MemoryStorageService memStore = new MemoryStorageService(); memStore.initStore(); - aStore.init(memStore); + ISessionsStore sessionsStore = memStore.sessionsStore(); + aStore.init(sessionsStore); //subscribe a not active clientID1 to /topic Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); aStore.add(slashSub); - memStore.createNewSession("FAKE_CLI_ID_1", true).deactivate(); + sessionsStore.createNewSession("FAKE_CLI_ID_1", true).deactivate(); //subscribe an active clientID2 to /topic Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", "/topic", AbstractMessage.QOSType.MOST_ONE, false); 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 99109d36..05a46ff4 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 @@ -46,9 +46,11 @@ public class MemorySessionStore implements ISessionsStore { //maps clientID->[messageID*] private Map> m_secondPhaseStore = new HashMap<>(); + private Map> m_messageToGuids; private final IMessagesStore m_messagesStore; - public MemorySessionStore(IMessagesStore messagesStore) { + public MemorySessionStore(IMessagesStore messagesStore, Map> messageToGuids) { + m_messageToGuids = messageToGuids; this.m_messagesStore = messagesStore; } @@ -217,6 +219,8 @@ public class MemorySessionStore implements ISessionsStore { @Override public String mapToGuid(String clientID, int messageID) { - return ((MemoryStorageService) m_messagesStore).mapToGuid(clientID, messageID); + HashMap guids = (HashMap) defaultGet(m_messageToGuids, + clientID, new HashMap()); + return guids.get(messageID); } }