Added QueueRepository implemented in memory and on H2

broker/src/main/java/io/moquette/broker/IQueueRepository.java

@@ -0,0 +1,8 @@
+package io.moquette.broker;
+
+import java.util.Queue;
+
+public interface IQueueRepository {
+
+    Queue<SessionRegistry.EnqueuedMessage> createQueue(String cli, boolean clean);
+}

broker/src/main/java/io/moquette/broker/MemoryQueueRepository.java

@@ -0,0 +1,12 @@
+package io.moquette.broker;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+public class MemoryQueueRepository implements IQueueRepository {
+
+    @Override
+    public Queue<SessionRegistry.EnqueuedMessage> createQueue(String cli, boolean clean) {
+        return new ConcurrentLinkedQueue<>();
+    }
+}

broker/src/main/java/io/moquette/broker/Server.java

@@ -161,18 +161,21 @@ public class Server {
         authorizatorPolicy = initializeAuthorizatorPolicy(authorizatorPolicy, config);
 
         final ISubscriptionsRepository subscriptionsRepository;
+        final IQueueRepository queueRepository;
         if (persistencePath != null && !persistencePath.isEmpty()) {
             LOG.trace("Configuring H2 subscriptions store to {}", persistencePath);
-            h2Builder = new H2Builder(config, scheduler);
-            subscriptionsRepository = h2Builder.initStore().subscriptionsRepository();
+            h2Builder = new H2Builder(config, scheduler).initStore();
+            subscriptionsRepository = h2Builder.subscriptionsRepository();
+            queueRepository = h2Builder.queueRepository();
         } else {
             LOG.trace("Configuring in-memory subscriptions store");
             subscriptionsRepository = new MemorySubscriptionsRepository();
+            queueRepository = new MemoryQueueRepository();
         }
 
         ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
         subscriptions.init(subscriptionsRepository);
-        SessionRegistry sessions = new SessionRegistry(subscriptions, interceptor);
+        SessionRegistry sessions = new SessionRegistry(subscriptions, queueRepository);
         dispatcher = new PostOffice(subscriptions, authorizatorPolicy, new MemoryRetainedRepository(), sessions,
                                     interceptor);
         final BrokerConfiguration brokerConfig = new BrokerConfiguration(config);

broker/src/main/java/io/moquette/broker/Session.java

@@ -97,10 +97,8 @@ class Session {
     private final AtomicInteger inflightSlots = new AtomicInteger(INFLIGHT_WINDOW_SIZE); // this should be configurable
 
     Session(String clientId, boolean clean, Will will, Queue<SessionRegistry.EnqueuedMessage> sessionQueue) {
-        this.clientId = clientId;
-        this.clean = clean;
+        this(clean, clientId, sessionQueue);
         this.will = will;
-        this.sessionQueue = sessionQueue;
     }
 
     Session(boolean clean, String clientId, Queue<SessionRegistry.EnqueuedMessage> sessionQueue) {

broker/src/main/java/io/moquette/broker/SessionRegistry.java

@@ -16,7 +16,6 @@
 package io.moquette.broker;
 
 import io.moquette.broker.Session.SessionStatus;
-import io.moquette.interception.BrokerInterceptor;
 import io.moquette.broker.subscriptions.ISubscriptionsDirectory;
 import io.moquette.broker.subscriptions.Subscription;
 import io.moquette.broker.subscriptions.Topic;
@@ -32,9 +31,9 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ConcurrentMap;
 
-class SessionRegistry {
+public class SessionRegistry {
 
-    abstract static class EnqueuedMessage {
+    public abstract static class EnqueuedMessage {
     }
 
     static class PublishedMessage extends EnqueuedMessage {
@@ -61,10 +60,12 @@ class SessionRegistry {
 
     private final ConcurrentMap<String, Session> pool = new ConcurrentHashMap<>();
     private final ISubscriptionsDirectory subscriptionsDirectory;
+    private final IQueueRepository queueRepository;
     private final ConcurrentMap<String, Queue<SessionRegistry.EnqueuedMessage>> queues = new ConcurrentHashMap<>();
 
-    SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory, BrokerInterceptor m_interceptor) {
+    SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory, IQueueRepository queueRepository) {
         this.subscriptionsDirectory = subscriptionsDirectory;
+        this.queueRepository = queueRepository;
     }
 
     void bindToSession(MQTTConnection mqttConnection, MqttConnectMessage msg, String clientId) {
@@ -178,9 +179,9 @@ class SessionRegistry {
     }
 
     private Session createNewSession(MQTTConnection mqttConnection, MqttConnectMessage msg, String clientId) {
-        final Queue<SessionRegistry.EnqueuedMessage> sessionQueue =
-                    queues.computeIfAbsent(clientId, (String cli) -> new ConcurrentLinkedQueue<>());
         final boolean clean = msg.variableHeader().isCleanSession();
+        final Queue<SessionRegistry.EnqueuedMessage> sessionQueue =
+                    queues.computeIfAbsent(clientId, (String cli) -> queueRepository.createQueue(cli, clean));
         final Session newSession;
         if (msg.variableHeader().isWillFlag()) {
             final Session.Will will = createWill(msg);

broker/src/main/java/io/moquette/persistence/H2Builder.java

@@ -1,6 +1,7 @@
 package io.moquette.persistence;
 
 import io.moquette.BrokerConstants;
+import io.moquette.broker.IQueueRepository;
 import io.moquette.broker.ISubscriptionsRepository;
 import io.moquette.broker.config.IConfig;
 import org.h2.mvstore.MVStore;
@@ -52,4 +53,8 @@ public class H2Builder {
     public void closeStore() {
         mvStore.close();
     }
+
+    public IQueueRepository queueRepository() {
+        return new H2QueueRepository(mvStore);
+    }
 }

broker/src/main/java/io/moquette/persistence/H2PersistentQueue.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2012-2018 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 io.moquette.persistence;
+
+import org.h2.mvstore.MVMap;
+import org.h2.mvstore.MVStore;
+
+import java.util.AbstractQueue;
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicLong;
+
+class H2PersistentQueue<T> extends AbstractQueue<T> {
+
+    private final MVMap<Long, T> queueMap;
+    private final MVMap<String, Long> metadataMap;
+    private final AtomicLong head;
+    private final AtomicLong tail;
+
+    H2PersistentQueue(MVStore store, String queueName) {
+        if (queueName == null || queueName.isEmpty()) {
+            throw new IllegalArgumentException("queueName parameter can't be empty or null");
+        }
+        this.queueMap = store.openMap("queue_" + queueName);
+        this.metadataMap = store.openMap("queue_" + queueName + "_meta");
+
+        //setup head index
+        long headIdx = 0L;
+        if (this.metadataMap.containsKey("head")) {
+            headIdx = this.metadataMap.get("head");
+        } else {
+            this.metadataMap.put("head", headIdx);
+        }
+        this.head = new AtomicLong(headIdx);
+
+        //setup tail index
+        long tailIdx = 0L;
+        if (this.metadataMap.containsKey("tail")) {
+            tailIdx = this.metadataMap.get("tail");
+        } else {
+            this.metadataMap.put("tail", tailIdx);
+        }
+        this.tail = new AtomicLong(tailIdx);
+    }
+
+    static void dropQueue(MVStore store, String queueName) {
+        store.removeMap(store.openMap("queue_" + queueName));
+        store.removeMap(store.openMap("queue_" + queueName + "_meta"));
+    }
+
+    @Override
+    public Iterator<T> iterator() {
+        return null;
+    }
+
+    @Override
+    public int size() {
+        return this.head.intValue() - this.tail.intValue();
+    }
+
+    @Override
+    public boolean offer(T t) {
+        if (t == null) {
+            throw new NullPointerException("Inserted element can't be null");
+        }
+        final long nextHead = head.getAndIncrement();
+        this.queueMap.put(nextHead, t);
+        this.metadataMap.put("head", nextHead + 1);
+        return true;
+    }
+
+    @Override
+    public T poll() {
+        if (head.equals(tail)) {
+            return null;
+        }
+        final long nextTail = tail.getAndIncrement();
+        final T tail = this.queueMap.get(nextTail);
+        queueMap.remove(nextTail);
+        this.metadataMap.put("tail", nextTail + 1);
+        return tail;
+    }
+
+    @Override
+    public T peek() {
+        if (head.equals(tail)) {
+            return null;
+        }
+        return this.queueMap.get(tail.get());
+    }
+
+}

broker/src/main/java/io/moquette/persistence/H2QueueRepository.java

@@ -0,0 +1,25 @@
+package io.moquette.persistence;
+
+import io.moquette.broker.IQueueRepository;
+import io.moquette.broker.SessionRegistry;
+import org.h2.mvstore.MVStore;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+public class H2QueueRepository implements IQueueRepository {
+
+    private MVStore mvStore;
+
+    public H2QueueRepository(MVStore mvStore) {
+        this.mvStore = mvStore;
+    }
+
+    @Override
+    public Queue<SessionRegistry.EnqueuedMessage> createQueue(String cli, boolean clean) {
+        if (!clean) {
+            return new H2PersistentQueue<>(mvStore, cli);
+        }
+        return new ConcurrentLinkedQueue<>();
+    }
+}

broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java

@@ -52,6 +52,7 @@ public class MQTTConnectionConnectTest {
     private static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false);
     private IAuthenticator mockAuthenticator;
     private PostOffice postOffice;
+    private MemoryQueueRepository queueRepository;
 
     @Before
     public void setUp() {
@@ -62,8 +63,9 @@ public class MQTTConnectionConnectTest {
         ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
+        queueRepository = new MemoryQueueRepository();
 
-        sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
+        sessionRegistry = new SessionRegistry(subscriptions, queueRepository);
         postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(),
                                     sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
 

broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java

@@ -45,6 +45,7 @@ public class MQTTConnectionPublishTest {
     private EmbeddedChannel channel;
     private SessionRegistry sessionRegistry;
     private MqttMessageBuilders.ConnectBuilder connMsg;
+    private MemoryQueueRepository queueRepository;
 
     @Before
     public void setUp() {
@@ -67,8 +68,9 @@ public class MQTTConnectionPublishTest {
         ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
+        queueRepository = new MemoryQueueRepository();
 
-        sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
+        sessionRegistry = new SessionRegistry(subscriptions, queueRepository);
         final PostOffice postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(),
                                                      new MemoryRetainedRepository(), sessionRegistry,
                                                      ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);

broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java

@@ -54,6 +54,7 @@ public class PostOfficeInternalPublishTest {
     private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID =
         new BrokerConfiguration(true, true, false);
     private MemoryRetainedRepository retainedRepository;
+    private MemoryQueueRepository queueRepository;
 
     @Before
     public void setUp() {
@@ -82,9 +83,9 @@ public class PostOfficeInternalPublishTest {
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
         retainedRepository = new MemoryRetainedRepository();
+        queueRepository = new MemoryQueueRepository();
 
-        SessionRegistry sessionRegistry = new SessionRegistry(subscriptions,
-                                                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
+        SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository);
         sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
         return sessionRegistry;

broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java

@@ -64,6 +64,7 @@ public class PostOfficePublishTest {
     static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID =
         new BrokerConfiguration(true, true, false);
     private MemoryRetainedRepository retainedRepository;
+    private MemoryQueueRepository queueRepository;
 
     @Before
     public void setUp() {
@@ -89,9 +90,9 @@ public class PostOfficePublishTest {
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
         retainedRepository = new MemoryRetainedRepository();
+        queueRepository = new MemoryQueueRepository();
 
-        SessionRegistry sessionRegistry = new SessionRegistry(subscriptions,
-                                                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
+        SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository);
         sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry,
                              ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
         return sessionRegistry;

broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java

@@ -65,6 +65,7 @@ public class PostOfficeSubscribeTest {
     private IAuthenticator mockAuthenticator;
     private SessionRegistry sessionRegistry;
     public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false);
+    private MemoryQueueRepository queueRepository;
 
     @Before
     public void setUp() {
@@ -87,8 +88,9 @@ public class PostOfficeSubscribeTest {
         subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
+        queueRepository = new MemoryQueueRepository();
 
-        sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
+        sessionRegistry = new SessionRegistry(subscriptions, queueRepository);
         sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(),
                              sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
     }

broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java

@@ -56,6 +56,7 @@ public class PostOfficeUnsubscribeTest {
     private IAuthenticator mockAuthenticator;
     private SessionRegistry sessionRegistry;
     public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false);
+    private MemoryQueueRepository queueRepository;
 
     @Before
     public void setUp() {
@@ -78,8 +79,9 @@ public class PostOfficeUnsubscribeTest {
         subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
+        queueRepository = new MemoryQueueRepository();
 
-        sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
+        sessionRegistry = new SessionRegistry(subscriptions, queueRepository);
         sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(),
                              sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
     }

broker/src/test/java/io/moquette/broker/SessionRegistryTest.java

@@ -47,6 +47,7 @@ public class SessionRegistryTest {
     private MqttMessageBuilders.ConnectBuilder connMsg;
     private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZEROBYTE_CLIENT_ID =
         new BrokerConfiguration(true, true, false);
+    private MemoryQueueRepository queueRepository;
 
     @Before
     public void setUp() {
@@ -67,8 +68,9 @@ public class SessionRegistryTest {
         ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
         ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
         subscriptions.init(subscriptionsRepository);
+        queueRepository = new MemoryQueueRepository();
 
-        sut = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
+        sut = new SessionRegistry(subscriptions, queueRepository);
         final PostOffice postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(),
                                                      new MemoryRetainedRepository(), sut,
                                                      ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);

broker/src/test/java/io/moquette/persistence/H2PersistentQueueTest.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright (c) 2012-2018 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 io.moquette.persistence;
+
+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;
+
+import static org.junit.Assert.*;
+
+public class H2PersistentQueueTest {
+
+    private MVStore mvStore;
+
+    @Before
+    public void setUp() {
+        this.mvStore = new MVStore.Builder()
+            .fileName(BrokerConstants.DEFAULT_PERSISTENT_PATH)
+            .autoCommitDisabled()
+            .open();
+    }
+
+    @After
+    public void tearDown() {
+        File dbFile = new File(BrokerConstants.DEFAULT_PERSISTENT_PATH);
+        if (dbFile.exists()) {
+            dbFile.delete();
+        }
+        assertFalse(dbFile.exists());
+    }
+
+    @Test
+    public void testAdd() {
+        H2PersistentQueue<String> sut = new H2PersistentQueue<>(this.mvStore, "test");
+
+        sut.add("Hello");
+        sut.add("world");
+
+        assertEquals("Hello", sut.peek());
+        assertEquals("Hello", sut.peek());
+        assertEquals("peek just return elements, doesn't remove them", 2, sut.size());
+    }
+
+    @Test
+    public void testPoll() {
+        H2PersistentQueue<String> sut = new H2PersistentQueue<>(this.mvStore, "test");
+        sut.add("Hello");
+        sut.add("world");
+
+        assertEquals("Hello", sut.poll());
+        assertEquals("world", sut.poll());
+        assertTrue("after poll 2 elements inserted before, should be empty", sut.isEmpty());
+    }
+
+    @Ignore
+    @Test
+    public void testPerformance() {
+        H2PersistentQueue<String> sut = new H2PersistentQueue<>(this.mvStore, "test");
+
+        int numIterations = 10000000;
+        for (int i = 0; i < numIterations; i++) {
+            sut.add("Hello");
+        }
+        mvStore.commit();
+
+        for (int i = 0; i < numIterations; i++) {
+            assertEquals("Hello", sut.poll());
+        }
+
+        assertTrue("should be empty", sut.isEmpty());
+    }
+
+    @Test
+    public void testReloadFromPersistedState() {
+        H2PersistentQueue<String> before = new H2PersistentQueue<>(this.mvStore, "test");
+        before.add("Hello");
+        before.add("crazy");
+        before.add("world");
+        assertEquals("Hello", before.poll());
+        this.mvStore.commit();
+        this.mvStore.close();
+
+        this.mvStore = new MVStore.Builder()
+            .fileName(BrokerConstants.DEFAULT_PERSISTENT_PATH)
+            .autoCommitDisabled()
+            .open();
+
+        //now reload the persisted state
+        H2PersistentQueue<String> after = new H2PersistentQueue<>(this.mvStore, "test");
+
+        assertEquals("crazy", after.poll());
+        assertEquals("world", after.poll());
+        assertTrue("should be empty", after.isEmpty());
+    }
+}