anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed the constraint on client queues to be BlockingQueue but simple Queue

Browse Source
This commit is contained in:
andsel
2017-06-20 21:55:59 +02:00
parent 1f3d39f4b1
commit 53ec8e5b60
10 changed files with 38 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/ClientSession.java
+3 -3
View File
@@ -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<IMessagesStore.StoredMessage> queue() {
LOG.info("Retrieving enqueued messages. ClientId={}", clientID);
public Queue<IMessagesStore.StoredMessage> queue() {
LOG.info("Retrieving enqueued messages. CId={}", clientID);
return this.m_sessionsStore.queue(clientID);
}
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+4 -4
View File
@@ -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<StoredMessage> queue(String clientID);
Queue<StoredMessage> queue(String clientID);
void dropQueue(String clientID);
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+5 -8
View File
@@ -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<IMessagesStore.StoredMessage> publishedEvents) {
List<IMessagesStore.StoredMessage> storedPublishes = new ArrayList<>();
publishedEvents.drainTo(storedPublishes);
for (IMessagesStore.StoredMessage pubEvt : storedPublishes) {
void publishStored(ClientSession clientSession, Queue<IMessagesStore.StoredMessage> 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());
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+3 -6
View File
@@ -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<StoredMessage> publishedEvents = clientSession.queue();
Queue<StoredMessage> 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;
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+1 -1
View File
@@ -372,7 +372,7 @@ public class ProtocolProcessorTest {
m_processor.processConnect(m_channel, connectMessage);
// Verify no messages are still stored
BlockingQueue<StoredMessage> messages = m_sessionStore.queue(FAKE_PUBLISHER_ID);
Queue<StoredMessage> messages = m_sessionStore.queue(FAKE_PUBLISHER_ID);
assertTrue(messages.isEmpty());
}
distribution/src/main/resources/moquette.conf
+2
View File
@@ -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:
h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java
+10 -5
View File
@@ -9,7 +9,7 @@ import java.util.concurrent.atomic.AtomicLong;
class H2PersistentQueue<T> extends AbstractQueue<T> {
private final MVMap<Long, T> mvMap;
private final MVMap<Long, T> queueMap;
private final MVMap<String, Long> metadataMap;
private final AtomicLong head;
private final AtomicLong tail;
@@ -18,7 +18,7 @@ class H2PersistentQueue<T> extends AbstractQueue<T> {
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<T> extends AbstractQueue<T> {
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;
@@ -56,7 +61,7 @@ class H2PersistentQueue<T> extends AbstractQueue<T> {
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<T> extends AbstractQueue<T> {
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<T> extends AbstractQueue<T> {
if (head.equals(tail)) {
return null;
}
return this.mvMap.get(tail.get());
return this.queueMap.get(tail.get());
}
}
h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java
+4 -45
View File
@@ -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<Integer> inFlightForClient = this.inFlightIds.get(clientID);
if (inFlightForClient == null) {
int nextPacketId = 1;
inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
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<T> extends AbstractQueue<T> {
private final MVMap<T, ?> map;
private final Cursor<T, ?> cursor;
H2BackedQueue(MVMap<T, ?> map) {
this.cursor = map.cursor(null);
this.map = map;
}
@Override
public Iterator<T> 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<StoredMessage> queue(String clientID) {
throw new RuntimeException("Not yet implemented");
public Queue<StoredMessage> 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
mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java
+2 -1
View File
@@ -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 {
mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java
+4 -5
View File
@@ -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<StoredMessage> queue(String clientID) {
LOG.info("Queuing pending message. ClientId={}, guid={}", clientID);
public Queue<StoredMessage> 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<Integer, StoredMessage> 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 " +