From 53ec8e5b6032a7979dace06c38feb6fa52f0f442 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 19 Jun 2017 22:40:34 +0200 Subject: [PATCH] Removed the constraint on client queues to be BlockingQueue but simple Queue --- .../java/io/moquette/spi/ClientSession.java | 6 +-- .../java/io/moquette/spi/ISessionsStore.java | 8 +-- .../spi/impl/InternalRepublisher.java | 13 ++--- .../moquette/spi/impl/ProtocolProcessor.java | 9 ++-- .../spi/impl/ProtocolProcessorTest.java | 2 +- distribution/src/main/resources/moquette.conf | 2 + .../persistence/h2/H2PersistentQueue.java | 15 ++++-- .../persistence/h2/H2SessionsStore.java | 49 ++----------------- .../mapdb/MapDBPersistentStore.java | 3 +- .../persistence/mapdb/MapDBSessionsStore.java | 9 ++-- 10 files changed, 38 insertions(+), 78 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index e7cf7824..a7b579d8 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -23,8 +23,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.HashSet; +import java.util.Queue; import java.util.Set; -import java.util.concurrent.BlockingQueue; /** * Model a Session like describe on page 25 of MQTT 3.1.1 specification: @@ -108,8 +108,8 @@ public class ClientSession { * * @return the list of messages to be delivered for client related to the session. */ - public BlockingQueue queue() { - LOG.info("Retrieving enqueued messages. ClientId={}", clientID); + public Queue queue() { + LOG.info("Retrieving enqueued messages. CId={}", clientID); return this.m_sessionsStore.queue(clientID); } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index f25842a5..dff911e9 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -19,7 +19,7 @@ package io.moquette.spi; import io.moquette.spi.IMessagesStore.StoredMessage; import java.util.Collection; -import java.util.concurrent.BlockingQueue; +import java.util.Queue; /** * Store used to handle the persistence of the subscriptions tree. @@ -79,13 +79,13 @@ public interface ISessionsStore { int nextPacketID(String clientID); /** - * List the guids for retained messages for the session + * List the published retained messages for the session * * @param clientID * the client ID owning the queue. - * @return the list of queue message UUIDs. + * @return the queue of messages. */ - BlockingQueue queue(String clientID); + Queue queue(String clientID); void dropQueue(String clientID); diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java index 7449ae74..071db32f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java +++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java @@ -22,10 +22,9 @@ import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.ArrayList; + import java.util.Collection; -import java.util.List; -import java.util.concurrent.BlockingQueue; +import java.util.Queue; class InternalRepublisher { @@ -54,11 +53,9 @@ class InternalRepublisher { } } - void publishStored(ClientSession clientSession, BlockingQueue publishedEvents) { - List storedPublishes = new ArrayList<>(); - publishedEvents.drainTo(storedPublishes); - - for (IMessagesStore.StoredMessage pubEvt : storedPublishes) { + void publishStored(ClientSession clientSession, Queue publishedEvents) { + IMessagesStore.StoredMessage pubEvt; + while ((pubEvt = publishedEvents.poll()) != null) { // put in flight zone LOG.debug("Adding message ot inflight zone. ClientId={}, guid={}, topic={}", clientSession.clientID, pubEvt.getGuid(), pubEvt.getTopic()); 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 97ddff34..cfee64f5 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -36,10 +36,7 @@ import io.netty.handler.timeout.IdleStateHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; -import java.util.UUID; +import java.util.*; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -452,9 +449,9 @@ public class ProtocolProcessor { */ private void republishStoredInSession(ClientSession clientSession) { LOG.info("Republishing stored publish events. CId={}", clientSession.clientID); - BlockingQueue publishedEvents = clientSession.queue(); + Queue publishedEvents = clientSession.queue(); if (publishedEvents.isEmpty()) { - LOG.info("There are no stored publish events to ClientId={}", clientSession.clientID); + LOG.info("There are no stored publish events to CId={}", clientSession.clientID); return; } 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 589761a3..59f787ca 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -372,7 +372,7 @@ public class ProtocolProcessorTest { m_processor.processConnect(m_channel, connectMessage); // Verify no messages are still stored - BlockingQueue messages = m_sessionStore.queue(FAKE_PUBLISHER_ID); + Queue messages = m_sessionStore.queue(FAKE_PUBLISHER_ID); assertTrue(messages.isEmpty()); } diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf index 87bad321..72aca208 100644 --- a/distribution/src/main/resources/moquette.conf +++ b/distribution/src/main/resources/moquette.conf @@ -41,6 +41,8 @@ host 0.0.0.0 # no physical persistence #********************************************************************* #persistent_store ./moquette_store.mapdb +#storage_class io.moquette.persistence.h2.H2PersistentStore +#storage_class io.moquette.persistence.mapdb.MapDBPersistentStore #********************************************************************* # acl_file: diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java index 3f661e4c..3b1fc895 100644 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java @@ -9,7 +9,7 @@ import java.util.concurrent.atomic.AtomicLong; class H2PersistentQueue extends AbstractQueue { - private final MVMap mvMap; + private final MVMap queueMap; private final MVMap metadataMap; private final AtomicLong head; private final AtomicLong tail; @@ -18,7 +18,7 @@ class H2PersistentQueue extends AbstractQueue { if (queueName == null || queueName.isEmpty()) { throw new IllegalArgumentException("queueName parameter can't be empty or null"); } - this.mvMap = store.openMap("queue_" + queueName); + this.queueMap = store.openMap("queue_" + queueName); this.metadataMap = store.openMap("queue_" + queueName + "_meta"); //setup head index @@ -40,6 +40,11 @@ class H2PersistentQueue extends AbstractQueue { 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 iterator() { return null; @@ -56,7 +61,7 @@ class H2PersistentQueue extends AbstractQueue { throw new NullPointerException("Inserted element can't be null"); } final long nextHead = head.getAndIncrement(); - this.mvMap.put(nextHead, t); + this.queueMap.put(nextHead, t); this.metadataMap.put("head", nextHead + 1); return true; } @@ -67,7 +72,7 @@ class H2PersistentQueue extends AbstractQueue { return null; } final long nextTail = tail.getAndIncrement(); - final T tail = this.mvMap.get(nextTail); + final T tail = this.queueMap.get(nextTail); this.metadataMap.put("tail", nextTail + 1); return tail; } @@ -77,7 +82,7 @@ class H2PersistentQueue extends AbstractQueue { if (head.equals(tail)) { return null; } - return this.mvMap.get(tail.get()); + return this.queueMap.get(tail.get()); } } diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java index c680926e..7a6840d1 100644 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java @@ -23,14 +23,12 @@ import io.moquette.spi.ISessionsStore; import io.moquette.spi.ISubscriptionsStore; import io.moquette.spi.impl.subscriptions.Subscription; 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.*; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -219,7 +217,7 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { Set inFlightForClient = this.inFlightIds.get(clientID); if (inFlightForClient == null) { int nextPacketId = 1; - inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap()); + inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<>()); inFlightForClient.add(nextPacketId); this.inFlightIds.put(clientID, inFlightForClient); return nextPacketId; @@ -232,53 +230,14 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { return nextPacketId; } - - class H2BackedQueue extends AbstractQueue { - - private final MVMap map; - private final Cursor cursor; - - H2BackedQueue(MVMap map) { - this.cursor = map.cursor(null); - this.map = map; - } - - @Override - public Iterator iterator() { - return cursor; - } - - @Override - public int size() { - return map.size(); - } - - @Override - public boolean offer(T t) { - return false; - } - - @Override - public T poll() { - return null; - } - - @Override - public T peek() { - return null; - } - } - - //TODO @Override - public BlockingQueue queue(String clientID) { - throw new RuntimeException("Not yet implemented"); + public Queue queue(String clientID) { + return new H2PersistentQueue<>(this.mvStore, clientID); } - //TODO @Override public void dropQueue(String clientID) { - throw new RuntimeException("Not yet implemented"); + H2PersistentQueue.dropQueue(this.mvStore, clientID); } @Override diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java index 7932d59b..55116ae9 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java @@ -25,6 +25,7 @@ import org.mapdb.DB; import org.mapdb.DBMaker; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.io.File; import java.io.IOException; import java.util.concurrent.ScheduledExecutorService; @@ -108,7 +109,7 @@ public class MapDBPersistentStore implements IStore { this.m_db.close(); LOG.info("Stopping MapDB commit tasks"); - //TODO th scheduler must be stopped by the owning (the instance of Server) + //TODO the scheduler must be stopped by the owning (the instance of Server) //invalidate the added task this.m_scheduler.shutdown(); try { diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java index 21aa99b3..d68cd606 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java @@ -28,7 +28,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -235,20 +234,20 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { } @Override - public BlockingQueue queue(String clientID) { - LOG.info("Queuing pending message. ClientId={}, guid={}", clientID); + public Queue queue(String clientID) { + LOG.info("Queuing pending message. CId={}, guid={}", clientID); return this.m_db.getQueue(clientID); } @Override public void dropQueue(String clientID) { - LOG.info("Removing pending messages. ClientId={}", clientID); + LOG.info("Removing pending messages. CId={}", clientID); this.m_db.delete(clientID); } @Override public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg) { - LOG.debug("Moving inflight message to 2nd phase ack state. ClientId={}, messageID={}", clientID, messageID); + LOG.debug("Moving inflight message to 2nd phase ack state. CId={}, messageID={}", clientID, messageID); ConcurrentMap m = this.m_secondPhaseStore.get(clientID); if (m == null) { String error = String.format("Can't find the inFlight record for client <%s> during the second phase of " +