diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 12091938..f693de53 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -59,4 +59,5 @@ public class BrokerConstants { public static final String NETTY_TCP_NODELAY_PROPERTY_NAME = "netty.tcp_nodelay"; public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive"; public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds"; + public static final String STORAGE_CLASS_NAME = "storage_class"; } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index b4b002c1..ba82030d 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/broker/src/main/java/io/moquette/server/Server.java @@ -329,4 +329,7 @@ public class Server { return m_processorBootstrapper.getConnectionDescriptors(); } + public ProtocolProcessor getProcessor() { + return m_processor; + } } diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 63efdfe8..c3f12ab0 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -188,7 +188,7 @@ public class ClientSession { } public IMessagesStore.StoredMessage storedMessage(int messageID) { - final MessageGUID guid = m_sessionsStore.mapToGuid(clientID, messageID); + final MessageGUID guid = messagesStore.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 8ef8039f..202beaec 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -142,4 +142,6 @@ public interface IMessagesStore { void cleanRetained(Topic topic); int getPendingPublishMessages(String clientID); + + MessageGUID mapToGuid(String clientID, int messageID); } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 00845313..6d795e9b 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -183,8 +183,6 @@ public interface ISessionsStore { */ MessageGUID secondPhaseAcknowledged(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/IStore.java b/broker/src/main/java/io/moquette/spi/IStore.java new file mode 100644 index 00000000..21a69376 --- /dev/null +++ b/broker/src/main/java/io/moquette/spi/IStore.java @@ -0,0 +1,12 @@ +package io.moquette.spi; + + +public interface IStore { + + IMessagesStore messagesStore(); + + ISessionsStore sessionsStore(); + + void close(); + +} 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 468245ae..4b2ce259 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -136,8 +136,11 @@ public class ProtocolProcessor { private boolean allowAnonymous; private boolean allowZeroByteClientId; private IAuthorizator m_authorizator; + private IMessagesStore m_messagesStore; + private ISessionsStore m_sessionsStore; + private IAuthenticator m_authenticator; private BrokerInterceptor m_interceptor; private String m_server_port; @@ -1063,4 +1066,12 @@ public class ProtocolProcessor { public void removeInterceptHandler(InterceptHandler interceptHandler) { this.m_interceptor.removeInterceptHandler(interceptHandler); } + + public IMessagesStore getMessagesStore() { + return m_messagesStore; + } + + public ISessionsStore getSessionsStore() { + return m_sessionsStore; + } } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 279b0b94..28453004 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -24,6 +24,7 @@ import io.moquette.server.config.IConfig; import io.moquette.server.config.IResourceLoader; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.IStore; import io.moquette.spi.impl.security.ACLFileParser; import io.moquette.spi.impl.security.AcceptAllAuthenticator; import io.moquette.spi.impl.security.DenyAllAuthorizator; @@ -53,10 +54,10 @@ public class ProtocolProcessorBootstrapper { private SubscriptionsStore subscriptions; - private MapDBPersistentStore m_mapStorage; - private ISessionsStore m_sessionsStore; + private Runnable storeShutdown = null; + private BrokerInterceptor m_interceptor; private final ProtocolProcessor m_processor = new ProtocolProcessor(); @@ -89,11 +90,31 @@ public class ProtocolProcessorBootstrapper { IAuthenticator authenticator, IAuthorizator authorizator, Server server) { subscriptions = new SubscriptionsStore(); + IMessagesStore messagesStore; LOG.info("Initializing messages and sessions stores..."); - m_mapStorage = new MapDBPersistentStore(props); - m_mapStorage.initStore(); - IMessagesStore messagesStore = m_mapStorage.messagesStore(); - m_sessionsStore = m_mapStorage.sessionsStore(); + String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, ""); + if (storageClassName != null && !storageClassName.isEmpty()) { + final IStore store = loadClass(storageClassName, IStore.class, Server.class, server); + messagesStore = store.messagesStore(); + m_sessionsStore = store.sessionsStore(); + storeShutdown = new Runnable() { + @Override + public void run() { + store.close(); + } + }; + } else { + final MapDBPersistentStore m_mapStorage = new MapDBPersistentStore(props); + m_mapStorage.initStore(); + messagesStore = m_mapStorage.messagesStore(); + m_sessionsStore = m_mapStorage.sessionsStore(); + storeShutdown = new Runnable() { + @Override + public void run() { + m_mapStorage.close(); + } + }; + } LOG.info("Configuring message interceptors..."); @@ -258,12 +279,17 @@ public class ProtocolProcessorBootstrapper { return instance; } + public ISessionsStore getSessionsStore() { + return m_sessionsStore; + } + public List getSubscriptions() { return m_sessionsStore.getSubscriptions(); } public void shutdown() { - this.m_mapStorage.close(); + if (storeShutdown != null) + storeShutdown.run(); } public ConnectionDescriptorStore getConnectionDescriptors() { 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 44b5a53c..79dd0389 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -91,7 +91,7 @@ class MapDBMessagesStore implements IMessagesStore { storedMessage.getTopic()); m_persistentMessageStore.put(guid, storedMessage); ConcurrentMap messageIdToGuid = m_db - .getHashMap(MapDBSessionsStore.messageId2GuidsMapName(storedMessage.getClientID())); + .getHashMap(messageId2GuidsMapName(storedMessage.getClientID())); messageIdToGuid.put(storedMessage.getMessageID(), guid); return guid; } @@ -99,8 +99,7 @@ class MapDBMessagesStore implements IMessagesStore { @Override public void dropMessagesInSession(String clientID) { LOG.debug("Dropping stored messages. ClientId = {}.", clientID); - ConcurrentMap messageIdToGuid = m_db - .getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)); + ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); for (MessageGUID guid : messageIdToGuid.values()) { removeStoredMessage(guid); } @@ -136,7 +135,20 @@ class MapDBMessagesStore implements IMessagesStore { @Override public int getPendingPublishMessages(String clientID) { ConcurrentMap messageIdToGuidMap = m_db - .getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID)); + .getHashMap(messageId2GuidsMapName(clientID)); return messageIdToGuidMap.size(); } + + @Override + public MessageGUID mapToGuid(String clientID, int messageID) { + LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID); + ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); + MessageGUID result = messageIdToGuid.get(messageID); + LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result); + return result; + } + + static String messageId2GuidsMapName(String clientID) { + return "guidsMapping_" + clientID; + } } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java index e9c8fe77..6a568e92 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java @@ -19,6 +19,7 @@ package io.moquette.spi.persistence; import io.moquette.server.config.IConfig; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.IStore; import org.mapdb.DB; import org.mapdb.DBMaker; import org.slf4j.Logger; @@ -35,7 +36,7 @@ import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME; /** * MapDB main persistence implementation */ -public class MapDBPersistentStore { +public class MapDBPersistentStore implements IStore { /** * This is a DTO used to persist minimal status (clean session and activation status) of a @@ -58,8 +59,8 @@ public class MapDBPersistentStore { private final int m_autosaveInterval; // in seconds protected final ScheduledExecutorService m_scheduler = Executors.newScheduledThreadPool(1); - private MapDBMessagesStore m_messageStore; - private MapDBSessionsStore m_sessionsStore; + private IMessagesStore m_messageStore; + private ISessionsStore m_sessionsStore; public MapDBPersistentStore(IConfig props) { this.m_storePath = props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, ""); @@ -71,10 +72,12 @@ public class MapDBPersistentStore { * * @return the message store instance. */ + @Override public IMessagesStore messagesStore() { return m_messageStore; } + @Override public ISessionsStore sessionsStore() { return m_sessionsStore; } @@ -121,6 +124,7 @@ public class MapDBPersistentStore { m_sessionsStore.initStore(); } + @Override public void close() { if (this.m_db.isClosed()) { LOG.warn("The MapDB store is already closed. Nothing will be done."); 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 b748eb7e..e1318041 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -17,6 +17,7 @@ package io.moquette.spi.persistence; 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; @@ -50,9 +51,9 @@ class MapDBSessionsStore implements ISessionsStore { private ConcurrentMap> m_secondPhaseStore; private final DB m_db; - private final MapDBMessagesStore m_messagesStore; + private final IMessagesStore m_messagesStore; - MapDBSessionsStore(DB db, MapDBMessagesStore messagesStore) { + MapDBSessionsStore(DB db, IMessagesStore messagesStore) { m_db = db; m_messagesStore = messagesStore; } @@ -296,19 +297,6 @@ class MapDBSessionsStore implements ISessionsStore { return guid; } - @Override - public MessageGUID mapToGuid(String clientID, int messageID) { - LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID); - ConcurrentMap messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID)); - MessageGUID result = messageIdToGuid.get(messageID); - LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result); - return result; - } - - static String messageId2GuidsMapName(String clientID) { - return "guidsMapping_" + clientID; - } - @Override public StoredMessage getInflightMessage(String clientID, int messageID) { LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID); diff --git a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java index f875d9b8..1b05c885 100644 --- a/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java +++ b/broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java @@ -20,6 +20,9 @@ import io.moquette.BrokerConstants; import io.moquette.server.config.IConfig; import io.moquette.server.config.MemoryConfig; import io.moquette.spi.impl.subscriptions.Topic; +import io.moquette.spi.impl.MemoryMessagesStore; +import io.moquette.spi.impl.MemoryStorageService; +import io.moquette.spi.persistence.MemorySessionStore; import io.moquette.spi.security.IAuthenticator; import io.moquette.spi.security.IAuthorizator; import org.junit.After; @@ -64,6 +67,16 @@ public class ConfigurationClassLoaderTest implements IAuthenticator, IAuthorizat assertTrue(true); } + @Test + public void loadStorage() throws Exception { + Properties props = new Properties(IntegrationUtils.prepareTestProperties()); + props.setProperty(BrokerConstants.STORAGE_CLASS_NAME, MemoryStorageService.class.getName()); + + startServer(props); + assertTrue(m_server.getProcessor().getMessagesStore() instanceof MemoryMessagesStore); + assertTrue(m_server.getProcessor().getSessionsStore() instanceof MemorySessionStore); + } + @Override public boolean checkValid(String clientID, String username, byte[] password) { return true; diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java index 35ad1852..2f2505f3 100644 --- a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java +++ b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java @@ -22,7 +22,6 @@ public class ClientSessionTest { public void setUp() { store = new SubscriptionsStore(); MemoryStorageService storageService = new MemoryStorageService(); - storageService.initStore(); this.sessionsStore = storageService.sessionsStore(); store.init(sessionsStore); 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 2e838a96..2921b8fe 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java @@ -34,10 +34,9 @@ public class MemoryMessagesStore implements IMessagesStore { private Map m_retainedStore = new HashMap<>(); private Map m_persistentMessageStore = new HashMap<>(); - private Map> m_messageToGuids; + private Map> m_messageToGuids = new HashMap<>(); - MemoryMessagesStore(Map> messageToGuids) { - m_messageToGuids = messageToGuids; + MemoryMessagesStore() { } @Override @@ -109,4 +108,11 @@ public class MemoryMessagesStore implements IMessagesStore { else return messageToGuids.size(); } + + @Override + public MessageGUID mapToGuid(String clientID, int messageID) { + HashMap guids = (HashMap) Utils + .defaultGet(m_messageToGuids, clientID, new HashMap()); + return guids.get(messageID); + } } 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 0d2627ca..efb37241 100644 --- a/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java @@ -18,28 +18,32 @@ package io.moquette.spi.impl; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.MessageGUID; +import io.moquette.spi.IStore; import io.moquette.spi.persistence.MemorySessionStore; -import java.util.HashMap; -import java.util.Map; -public class MemoryStorageService { +public class MemoryStorageService implements IStore { private MemorySessionStore m_sessionsStore; private MemoryMessagesStore m_messagesStore; - public void initStore() { - Map> messageToGuids = new HashMap<>(); - m_messagesStore = new MemoryMessagesStore(messageToGuids); - m_sessionsStore = new MemorySessionStore(m_messagesStore, messageToGuids); + public MemoryStorageService() { + m_messagesStore = new MemoryMessagesStore(); + m_sessionsStore = new MemorySessionStore(m_messagesStore); + m_messagesStore.initStore(); + m_sessionsStore.initStore(); } + @Override public IMessagesStore messagesStore() { return m_messagesStore; } + @Override public ISessionsStore sessionsStore() { return m_sessionsStore; } + @Override + public void close() { + } } 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 d5bf38bb..925d0815 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -84,7 +84,6 @@ public class ProtocolProcessorTest { // sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); MemoryStorageService memStorage = new MemoryStorageService(); - memStorage.initStore(); m_messagesStore = memStorage.messagesStore(); m_sessionStore = memStorage.sessionsStore(); // m_messagesStore.initStore(); @@ -128,7 +127,6 @@ public class ProtocolProcessorTest { // simulate a connect that register a clientID to an IoSession MemoryStorageService storageService = new MemoryStorageService(); - storageService.initStore(); subs.init(storageService.sessionsStore()); m_processor.init( subs, @@ -177,7 +175,6 @@ public class ProtocolProcessorTest { // simulate a connect that register a clientID to an IoSession MemoryStorageService storageService = new MemoryStorageService(); - storageService.initStore(); subs.init(storageService.sessionsStore()); m_processor.init( subs, @@ -353,7 +350,6 @@ public class ProtocolProcessorTest { } }; MemoryStorageService storageService = new MemoryStorageService(); - storageService.initStore(); subs.init(storageService.sessionsStore()); // simulate a connect that register a clientID to an IoSession diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 18e35e3f..ce6bdeb1 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -66,7 +66,6 @@ public class ProtocolProcessor_CONNECT_Test { // sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); MemoryStorageService memStorage = new MemoryStorageService(); - memStorage.initStore(); m_messagesStore = memStorage.messagesStore(); m_sessionStore = memStorage.sessionsStore(); // m_messagesStore.initStore(); diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index 7d3ea50b..4617cbc9 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -43,7 +43,6 @@ public class SubscriptionsStoreTest { public void setUp() throws IOException { store = new SubscriptionsStore(); MemoryStorageService storageService = new MemoryStorageService(); - storageService.initStore(); this.sessionsStore = storageService.sessionsStore(); store.init(sessionsStore); } @@ -209,7 +208,6 @@ public class SubscriptionsStoreTest { Topic topic = new Topic(t); store = new SubscriptionsStore(); MemoryStorageService memStore = new MemoryStorageService(); - memStore.initStore(); ISessionsStore aSessionsStore = memStore.sessionsStore(); store.init(aSessionsStore); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, MqttQoS.AT_MOST_ONCE); @@ -223,7 +221,6 @@ public class SubscriptionsStoreTest { Topic topic = new Topic(t); store = new SubscriptionsStore(); MemoryStorageService memStore = new MemoryStorageService(); - memStore.initStore(); store.init(memStore.sessionsStore()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, MqttQoS.AT_MOST_ONCE); sessionsStore.addNewSubscription(sub); @@ -276,7 +273,6 @@ public class SubscriptionsStoreTest { public void removeSubscription_withDifferentClients_subscribedSameTopic() { SubscriptionsStore aStore = new SubscriptionsStore(); MemoryStorageService memStore = new MemoryStorageService(); - memStore.initStore(); ISessionsStore sessionsStore = memStore.sessionsStore(); aStore.init(sessionsStore); // subscribe a not active clientID1 to /topic 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 3d6df059..61608aef 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -52,11 +52,9 @@ public class MemorySessionStore implements ISessionsStore { // maps clientID->[MessageId -> guid] private Map> m_secondPhaseStore = new HashMap<>(); - private Map> m_messageToGuids; private final IMessagesStore m_messagesStore; - public MemorySessionStore(IMessagesStore messagesStore, Map> messageToGuids) { - m_messageToGuids = messageToGuids; + public MemorySessionStore(IMessagesStore messagesStore) { this.m_messagesStore = messagesStore; } @@ -235,13 +233,6 @@ public class MemorySessionStore implements ISessionsStore { return guid; } - @Override - public MessageGUID mapToGuid(String clientID, int messageID) { - HashMap guids = (HashMap) Utils - .defaultGet(m_messageToGuids, clientID, new HashMap()); - return guids.get(messageID); - } - @Override public StoredMessage getInflightMessage(String clientID, int messageID) { return null; diff --git a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java index 33646927..79a5196a 100644 --- a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java +++ b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java @@ -20,7 +20,6 @@ public class PerformanceTest { public static void main(String[] args) throws IOException, InterruptedException { SubscriptionsStore store = new SubscriptionsStore(); MemoryStorageService memStore = new MemoryStorageService(); - memStore.initStore(); ISessionsStore aSessionsStore = memStore.sessionsStore(); store.init(aSessionsStore);