diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java index 126494f8..d10b2bd2 100644 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2MessagesStore.java @@ -13,18 +13,17 @@ * * You may elect to redistribute this code under either of these licenses. */ + package io.moquette.persistence.h2; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; -import io.moquette.spi.MessageGUID; import io.moquette.spi.impl.subscriptions.Topic; import org.h2.mvstore.Cursor; import org.h2.mvstore.MVMap; import org.h2.mvstore.MVStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import java.util.*; class H2MessagesStore implements IMessagesStore { @@ -33,12 +32,7 @@ class H2MessagesStore implements IMessagesStore { private final MVStore mvStore; - // maps clientID -> guid - private MVMap retainedStore; - // maps guid to message, it's message store - private MVMap persistentMessageStore; - - private MVMap new_retainedStore; + private MVMap retainedStore; public H2MessagesStore(MVStore mvStore) { this.mvStore = mvStore; @@ -47,8 +41,6 @@ class H2MessagesStore implements IMessagesStore { @Override public void initStore() { retainedStore = mvStore.openMap("retained"); - persistentMessageStore = mvStore.openMap("oldPersistedMessages"); - new_retainedStore = mvStore.openMap("persistedMessages"); LOG.info("Initialized message H2 store"); } @@ -56,9 +48,9 @@ class H2MessagesStore implements IMessagesStore { public void storeRetained(Topic topic, StoredMessage storedMessage) { LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID()); if (storedMessage.getClientID() == null) { - throw new IllegalArgumentException( "Message to be persisted must have a not null client ID"); + throw new IllegalArgumentException("Message to be persisted must have a not null client ID"); } - new_retainedStore.put(topic, storedMessage); + retainedStore.put(topic, storedMessage); } @Override @@ -66,19 +58,14 @@ class H2MessagesStore implements IMessagesStore { LOG.debug("Scanning retained messages"); List results = new ArrayList<>(); - Cursor mapCursor = retainedStore.cursor(null); + Cursor mapCursor = retainedStore.cursor(null); while (mapCursor.hasNext()) { - final MessageGUID guid = mapCursor.getValue(); - final Topic topic = mapCursor.getKey(); - StoredMessage storedMsg = persistentMessageStore.get(guid); - if (condition.match(topic)) { - results.add(storedMsg); - } + Topic topic = mapCursor.next(); + if (topic != null && condition.match(topic)) + results.add(mapCursor.getValue()); } - if (LOG.isTraceEnabled()) { - LOG.trace("Retained messages have been scanned matchingMessages={}", results); - } + LOG.trace("Retained messages have been scanned matchingMessages={}", results); return results; } diff --git a/h2_storage/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java b/h2_storage/src/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java similarity index 100% rename from h2_storage/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java rename to h2_storage/src/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java diff --git a/h2_storage/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java b/h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java similarity index 98% rename from h2_storage/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java rename to h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java index 4304b491..1e4b911b 100644 --- a/h2_storage/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java +++ b/h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java @@ -4,6 +4,7 @@ import io.moquette.BrokerConstants; import org.h2.mvstore.MVStore; import org.junit.After; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import java.io.File; @@ -54,6 +55,7 @@ public class H2PersistentQueueTest { assertTrue("after poll 2 elements inserted before, should be empty", sut.isEmpty()); } + @Ignore @Test public void testPerformance() { H2PersistentQueue sut = new H2PersistentQueue<>(this.mvStore, "test");