From c793ded63e995f837d87cee7725054a5832f78fb Mon Sep 17 00:00:00 2001 From: andsel Date: Sat, 24 Oct 2015 16:26:43 +0200 Subject: [PATCH] Introduced 'fat' ClientSession class responsible to model the concept of an MQTT client session --- .../eclipse/moquette/spi/ClientSession.java | 59 +++++++++ .../eclipse/moquette/spi/IMessagesStore.java | 2 +- .../eclipse/moquette/spi/ISessionsStore.java | 7 +- .../moquette/spi/impl/ProtocolProcessor.java | 21 +-- .../spi/persistence/MapDBPersistentStore.java | 19 ++- .../server/ServerIntegrationFuseTest.java | 4 +- .../moquette/spi/impl/MemorySessionStore.java | 28 +++- .../spi/impl/MemoryStorageService.java | 124 +++++++----------- .../spi/impl/ProtocolProcessorTest.java | 19 ++- .../impl/ProtocolProcessor_CONNECT_Test.java | 3 +- .../subscriptions/SubscriptionsStoreTest.java | 4 +- 11 files changed, 184 insertions(+), 106 deletions(-) create mode 100644 broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java rename broker/src/{main => test}/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java (73%) diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java new file mode 100644 index 00000000..771f64e3 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java @@ -0,0 +1,59 @@ +/* + * 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; + +import org.eclipse.moquette.spi.impl.events.PublishEvent; + +import java.util.List; + +/** + * @author andrea + */ +public class ClientSession { + + public final String clientID; + + private final IMessagesStore messagesStore; + + public ClientSession(String clientID, IMessagesStore messagesStore) { + this.clientID = clientID; + this.messagesStore = messagesStore; + } + //List of client's subscriptions + //list of messages not acknowledged by client + //list of in-flight messages + + /** + * @return the list of messages to be delivered for client related to the session. + * */ + public List storedMessages() { + return messagesStore.listMessagesInSession(clientID); + } + + /** + * Remove a message previously stored for delivery. + * */ + public void removeDelivered(int messageID) { + messagesStore.removeMessageInSession(clientID, messageID); + } + + @Override + public String toString() { + return "ClientSession{" + + "clientID='" + clientID + '\'' + + '}'; + } +} \ No newline at end of file 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 841abc66..c3a89687 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -28,7 +28,7 @@ import java.util.List; */ public interface IMessagesStore { - public static class StoredMessage implements Serializable { + class StoredMessage implements Serializable { final AbstractMessage.QOSType m_qos; final byte[] m_payload; final String m_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 1c5e8b29..de9e6cc9 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java @@ -54,5 +54,10 @@ public interface ISessionsStore { */ boolean contains(String clientID); - void createNewSession(String clientID); + ClientSession createNewSession(String clientID); + + /** + * @return the session for the given clientID, null if not found. + * */ + ClientSession sessionForClient(String clientID); } 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 43174b56..10374516 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 @@ -20,6 +20,7 @@ import java.util.*; import java.util.concurrent.ConcurrentHashMap; 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.ISessionsStore; @@ -220,7 +221,9 @@ public class ProtocolProcessor { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED); - boolean isSessionAlreadyStored = m_sessionsStore.contains(msg.getClientID()); + + ClientSession clientSession = m_sessionsStore.sessionForClient(msg.getClientID()); + boolean isSessionAlreadyStored = clientSession != null; if (!msg.isCleanSession() && isSessionAlreadyStored) { okResp.setSessionPresent(true); } @@ -229,12 +232,12 @@ public class ProtocolProcessor { if (!isSessionAlreadyStored) { LOG.info("Create persistent session for clientID <{}>", msg.getClientID()); - m_sessionsStore.createNewSession(msg.getClientID()); + clientSession = m_sessionsStore.createNewSession(msg.getClientID()); } LOG.info("Connected client ID <{}> with clean session {}", msg.getClientID(), msg.isCleanSession()); if (!msg.isCleanSession()) { //force the republish of stored QoS1 and QoS2 - republishStoredInSession(msg.getClientID()); + republishStoredInSession(clientSession); } } @@ -248,19 +251,19 @@ public class ProtocolProcessor { /** * Republish QoS1 and QoS2 messages stored into the session for the clientID. * */ - private void republishStoredInSession(String clientID) { - LOG.trace("republishStoredInSession for client <{}>", clientID); - List publishedEvents = m_messagesStore.listMessagesInSession(clientID); + private void republishStoredInSession(ClientSession clientSession) { + LOG.trace("republishStoredInSession for client <{}>", clientSession); + List publishedEvents = clientSession.storedMessages(); if (publishedEvents.isEmpty()) { - LOG.info("No stored messages for client <{}>", clientID); + LOG.info("No stored messages for client <{}>", clientSession.clientID); return; } - LOG.info("republishing stored messages to client <{}>", clientID); + LOG.info("republishing stored messages to client <{}>", clientSession.clientID); for (PublishEvent pubEvt : publishedEvents) { sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID()); - m_messagesStore.removeMessageInSession(clientID, pubEvt.getMessageID()); + clientSession.removeDelivered(pubEvt.getMessageID()); } } 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 9063c619..07e63f87 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,6 +17,7 @@ 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; @@ -257,7 +258,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { Set subs = m_persistentSubscriptions.get(clientID); if (!subs.contains(newSubscription)) { - LOG.debug("updating clientID {} subscriptions set with new subscription", clientID); + 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) { @@ -300,14 +301,24 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } @Override - public void createNewSession(String clientID) { + public ClientSession createNewSession(String clientID) { LOG.debug("createNewSession for client <{}>", clientID); if (m_persistentSubscriptions.containsKey(clientID)) { - LOG.error("already exists a session for client <{}>", clientID); - return; + 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()); + return new ClientSession(clientID, this); + } + + @Override + public ClientSession sessionForClient(String clientID) { + if (!m_persistentSubscriptions.containsKey(clientID)) { + return null; + } + + return new ClientSession(clientID, this); } @Override diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java index cf3ff44c..0193491d 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java @@ -96,8 +96,8 @@ public class ServerIntegrationFuseTest { } @Test - public void checkWillTestmaentIsPublishedOnConnectionKill_noRetain() throws Exception { - LOG.info("checkWillTestmaentIsPublishedOnConnectionKill"); + public void checkWillTestamentIsPublishedOnConnectionKill_noRetain() throws Exception { + LOG.info("checkWillTestamentIsPublishedOnConnectionKill"); String willTestamentTopic = "/will/test"; String willTestamentMsg = "Bye bye"; diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java index 0648bbaa..581ff262 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java @@ -15,6 +15,8 @@ */ package org.eclipse.moquette.spi.impl; +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.slf4j.Logger; @@ -24,13 +26,19 @@ import java.util.*; /** * - * @author andrea + * @author andream_messagesStore */ public class MemorySessionStore implements ISessionsStore { private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class); private Map> m_persistentSubscriptions = new HashMap<>(); + private final IMessagesStore m_messagesStore; + + public MemorySessionStore(IMessagesStore messagesStore) { + this.m_messagesStore = messagesStore; + } + @Override public void removeSubscription(String topic, String clientID) { LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); @@ -82,12 +90,24 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public void createNewSession(String clientID) { + public ClientSession createNewSession(String clientID) { + LOG.debug("createNewSession for client <{}>", clientID); if (m_persistentSubscriptions.containsKey(clientID)) { - LOG.error("already exists a session for client <{}>", clientID); - return; + 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()); + return new ClientSession(clientID, m_messagesStore); + } + + @Override + public ClientSession sessionForClient(String clientID) { + if (!m_persistentSubscriptions.containsKey(clientID)) { + return null; + } + + return new ClientSession(clientID, m_messagesStore); } @Override diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java similarity index 73% rename from broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java rename to broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index de4d076c..60dfc9ac 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -18,6 +18,8 @@ package org.eclipse.moquette.spi.impl; import java.nio.ByteBuffer; import java.util.*; +import jdk.nashorn.internal.objects.NativeJava; +import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.impl.events.PublishEvent; @@ -34,17 +36,18 @@ import static org.eclipse.moquette.spi.impl.Utils.defaultGet; */ public class MemoryStorageService implements IMessagesStore, ISessionsStore { - private Map> m_persistentSubscriptions = new HashMap<>(); private Map m_retainedStore = new HashMap<>(); //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map) private Map> m_persistentMessageStore = new HashMap<>(); private Map m_inflightStore = new HashMap<>(); private Map> m_inflightIDs = new HashMap<>(); private Map m_qos2Store = new HashMap<>(); + private MemorySessionStore m_sessionsStore; private static final Logger LOG = LoggerFactory.getLogger(MemoryStorageService.class); - + public void initStore() { + m_sessionsStore = new MemorySessionStore(this); } @Override @@ -69,7 +72,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { public Collection searchMatching(IMatchingCondition condition) { LOG.debug("searchMatching scanning all retained messages, presents are {}", m_retainedStore.size()); - List results = new ArrayList(); + List results = new ArrayList<>(); for (Map.Entry entry : m_retainedStore.entrySet()) { StoredMessage storedMsg = entry.getValue(); @@ -84,13 +87,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { @Override public void storePublishForFuture(PublishEvent evt) { LOG.debug("storePublishForFuture store evt {}", evt); -// List storedEvents; String clientID = evt.getClientID(); -// if (!m_persistentMessageStore.containsKey(clientID)) { -// storedEvents = new ArrayList(); -// } else { -// storedEvents = m_persistentMessageStore.get(clientID); -// } List storedEvents = defaultGet(m_persistentMessageStore, clientID, new ArrayList()); storedEvents.add(evt); m_persistentMessageStore.put(clientID, storedEvents); @@ -143,7 +140,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } /** - * Return the next valid packetIdentifer for the given client session. + * Return the next valid packetIdentifier for the given client session. * */ @Override public int nextPacketID(String clientID) { @@ -161,73 +158,6 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { 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); - } - } - - @Override - public void addNewSubscription(Subscription newSubscription) { - final String clientID = newSubscription.getClientId(); - if (!m_persistentSubscriptions.containsKey(clientID)) { - m_persistentSubscriptions.put(clientID, new HashSet()); - } - - Set subs = m_persistentSubscriptions.get(clientID); - if (!subs.contains(newSubscription)) { - subs.add(newSubscription); - m_persistentSubscriptions.put(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); - } - - @Override - public boolean contains(String clientID) { - return m_persistentSubscriptions.containsKey(clientID); - } - - @Override - public void createNewSession(String clientID) { - if (m_persistentSubscriptions.containsKey(clientID)) { - LOG.error("already exists a session for client <{}>", clientID); - return; - } - m_persistentSubscriptions.put(clientID, new HashSet()); - } - - @Override - public List listAllSubscriptions() { - List allSubscriptions = new ArrayList(); - for (Map.Entry> entry : m_persistentSubscriptions.entrySet()) { - allSubscriptions.addAll(entry.getValue()); - } - return allSubscriptions; - } @Override public void close() { @@ -249,4 +179,44 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { public PublishEvent retrieveQoS2Message(String publishKey) { return m_qos2Store.get(publishKey); } + + @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) { + return m_sessionsStore.createNewSession(clientID); + } + + @Override + public ClientSession sessionForClient(String clientID) { + return m_sessionsStore.sessionForClient(clientID); + } } 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 cde49965..36d73182 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 @@ -75,6 +75,7 @@ public class ProtocolProcessorTest { //sleep to let the messaging batch processor to process the initEvent Thread.sleep(300); MemoryStorageService memStorage = new MemoryStorageService(); + memStorage.initStore(); m_storageService = memStorage; m_sessionStore = memStorage; //m_storageService.initStore(); @@ -84,7 +85,7 @@ public class ProtocolProcessorTest { m_mockAuthenticator = new MockAuthenticator(users); subscriptions = new SubscriptionsStore(); - subscriptions.init(new MemoryStorageService()); + subscriptions.init(memStorage); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); @@ -109,7 +110,9 @@ public class ProtocolProcessorTest { }; //simulate a connect that register a clientID to an IoSession - subs.init(new MemoryStorageService()); + MemoryStorageService storageService = new MemoryStorageService(); + storageService.initStore(); + subs.init(storageService); m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProtocolVersion((byte) 3); @@ -153,7 +156,9 @@ public class ProtocolProcessorTest { }; //simulate a connect that register a clientID to an IoSession - subs.init(new MemoryStorageService()); + MemoryStorageService storageService = new MemoryStorageService(); + storageService.initStore(); + subs.init(storageService); m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); @@ -306,8 +311,10 @@ public class ProtocolProcessorTest { } } }; - subs.init(new MemoryStorageService()); - + MemoryStorageService storageService = new MemoryStorageService(); + storageService.initStore(); + subs.init(storageService); + //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); ConnectMessage connectMessage = new ConnectMessage(); @@ -457,7 +464,7 @@ public class ProtocolProcessorTest { Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE, false); Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE, false); SubscriptionsStore subscriptions = new SubscriptionsStore(); - subscriptions.init(new MemorySessionStore()); + subscriptions.init(new MemorySessionStore(memoryMessageStore)); subscriptions.add(subQos1); subscriptions.add(subQos2); 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 2a51a99e..e0b5a47d 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 @@ -59,6 +59,7 @@ 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_storageService = memStorage; m_sessionStore = memStorage; //m_storageService.initStore(); @@ -68,7 +69,7 @@ public class ProtocolProcessor_CONNECT_Test { m_mockAuthenticator = new MockAuthenticator(users); subscriptions = new SubscriptionsStore(); - subscriptions.init(new MemoryStorageService()); + subscriptions.init(memStorage); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); 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 be5295c5..d374aec4 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 @@ -41,7 +41,9 @@ public class SubscriptionsStoreTest { @Before public void setUp() throws IOException { store = new SubscriptionsStore(); - store.init(new MemoryStorageService()); + MemoryStorageService storageService = new MemoryStorageService(); + storageService.initStore(); + store.init(storageService); } @Test