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

Removed old 'map based queues' to use persistent BlockingQueues from MapDB

Browse Source
This commit is contained in:
andsel
2017-01-09 21:53:26 +01:00
parent 3f0ab75efb
commit 081631242f
13 changed files with 64 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/spi/ClientSession.java
+14 -35
View File
@@ -16,19 +16,14 @@
package io.moquette.spi;
import io.moquette.parser.proto.messages.AbstractMessage;
import io.moquette.parser.proto.messages.PublishMessage;
import io.moquette.server.Constants;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
import io.moquette.spi.ISessionsStore.ClientTopicCouple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
@@ -61,7 +56,7 @@ public class ClientSession {
private volatile boolean cleanSession;
private BlockingQueue<AbstractMessage> m_queueToPublish = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE);
// private BlockingQueue<AbstractMessage> m_queueToPublish = new ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE);
public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore,
boolean cleanSession) {
@@ -72,21 +67,14 @@ public class ClientSession {
}
/**
* Return the list of persisted publishes for the given clientID.
* For QoS1 and QoS2 with clean session flag, this method return the list of
* missed publish events while the client was disconnected.
*
* @return the list of messages to be delivered for client related to the session.
* */
public List<IMessagesStore.StoredMessage> storedMessages() {
//read all messages from enqueued store
Collection<MessageGUID> guids = this.m_sessionsStore.enqueued(clientID);
return messagesStore.listMessagesInSession(guids);
}
/**
* Remove the messages stored in a cleanSession false.
*
* @param guid the guid of the message to remove from the queue.
*/
public void removeEnqueued(MessageGUID guid) {
this.m_sessionsStore.removeEnqueued(this.clientID, guid);
public BlockingQueue<IMessagesStore.StoredMessage> queue() {
return this.m_sessionsStore.queue(clientID);
}
@Override
@@ -173,8 +161,12 @@ public class ClientSession {
return messagesStore.getMessageByGuid(guid);
}
public void enqueueToDeliver(MessageGUID guid) {
this.m_sessionsStore.bindToDeliver(guid, this.clientID);
/**
* Enqueue a message to be sent to the client.
* @param message the message to enqueue.
* */
public void enqueue(IMessagesStore.StoredMessage message) {
this.m_sessionsStore.queue(this.clientID).add(message);
}
public IMessagesStore.StoredMessage storedMessage(int messageID) {
@@ -182,19 +174,6 @@ public class ClientSession {
return messagesStore.getMessageByGuid(guid);
}
/**
* Enqueue a message to be sent to the client.
* @param pubMessage the message to enqueue.
* @return false if the queue is full.
* */
public boolean enqueue(PublishMessage pubMessage) {
return m_queueToPublish.offer(pubMessage);
}
public AbstractMessage dequeue() {
return m_queueToPublish.poll();
}
public void moveInFlightToSecondPhaseAckWaiting(int messageID) {
m_sessionsStore.moveInFlightToSecondPhaseAckWaiting(this.clientID, messageID);
}
broker/src/main/java/io/moquette/spi/IMessagesStore.java
-9
View File
@@ -144,15 +144,6 @@ public interface IMessagesStore {
* */
MessageGUID storePublishForFuture(StoredMessage storedMessage);
/**
* Return the list of persisted publishes for the given clientID.
* For QoS1 and QoS2 with clean session flag, this method return the list of
* missed publish events while the client was disconnected.
* @param guids the list of of guid to use as search keys.
* @return the list of stored messages matching the passed keys.
*/
List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids);
void dropMessagesInSession(String clientID);
StoredMessage getMessageByGuid(MessageGUID guid);
broker/src/main/java/io/moquette/spi/ISessionsStore.java
+4 -16
View File
@@ -15,8 +15,8 @@
*/
package io.moquette.spi;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.impl.subscriptions.Subscription;
@@ -135,26 +135,14 @@ public interface ISessionsStore {
* */
int nextPacketID(String clientID);
/**
* Store the guid to be later published.
* @param guid the message guid to bind.
* @param clientID the clientID
* */
void bindToDeliver(MessageGUID guid, String clientID);
/**
* List the guids for retained messages for the session
* @param clientID the client ID owning the queue.
* @return the list of enqueued message UUIDs.
* @return the list of queue message UUIDs.
* */
Collection<MessageGUID> enqueued(String clientID);
BlockingQueue<StoredMessage> queue(String clientID);
/**
* Remove form the queue of stored messages for session.
* @param clientID the client ID owning the queue.
* @param guid the message guid to dequeue.
* */
void removeEnqueued(String clientID, MessageGUID guid);
void dropQueue(String clientID);
void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID);
broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+9 -3
View File
@@ -9,8 +9,11 @@ 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.Queue;
import java.util.concurrent.BlockingQueue;
class InternalRepublisher {
@@ -39,8 +42,11 @@ class InternalRepublisher {
}
}
void publishStored(ClientSession clientSession, List<IMessagesStore.StoredMessage> publishedEvents) {
for (IMessagesStore.StoredMessage pubEvt : publishedEvents) {
void publishStored(ClientSession clientSession, BlockingQueue<IMessagesStore.StoredMessage> publishedEvents) {
List<IMessagesStore.StoredMessage> storedPublishes = new ArrayList<>();
publishedEvents.drainTo(storedPublishes);
for (IMessagesStore.StoredMessage pubEvt : storedPublishes) {
//put in flight zone
LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID());
clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
@@ -61,7 +67,7 @@ class InternalRepublisher {
return createPublishForQos(storedMessage.getTopic(), storedMessage.getQos(), storedMessage.getMessage(), true);
}
private PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
public static PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
pubMessage.setTopicName(topic);
broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+2 -1
View File
@@ -48,6 +48,7 @@ class MessagesPublisher {
final ByteBuffer origMessage = pubMsg.getMessage();
//if QoS 1 or 2 store the message
//TODO perhaps this block is not needed.
MessageGUID guid = null;
if (publishingQos != AbstractMessage.QOSType.MOST_ONE) {
guid = m_messagesStore.storePublishForFuture(pubMsg);
@@ -76,7 +77,7 @@ class MessagesPublisher {
} else {
if (!targetSession.isCleanSession()) {
//store the message in targetSession queue to deliver
targetSession.enqueueToDeliver(guid);
targetSession.enqueue(pubMsg);
}
}
}
broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java
+2 -1
View File
@@ -10,6 +10,7 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.ConcurrentMap;
import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
import static io.moquette.spi.impl.ProtocolProcessor.asStoredMessage;
class PersistentQueueMessageSender {
@@ -48,7 +49,7 @@ class PersistentQueueMessageSender {
} else if (pubMessage.getQos() != MOST_ONE) {
//enqueue to the client session
LOG.debug("enqueue to client session");
clientsession.enqueue(pubMessage);
clientsession.enqueue(asStoredMessage(pubMessage));
}
}
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+11 -15
View File
@@ -17,9 +17,7 @@ package io.moquette.spi.impl;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import io.moquette.interception.InterceptHandler;
import io.moquette.parser.proto.messages.*;
@@ -36,6 +34,8 @@ import io.moquette.spi.impl.subscriptions.Subscription;
import static io.moquette.parser.netty.Utils.VERSION_3_1;
import static io.moquette.parser.netty.Utils.VERSION_3_1_1;
import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.parser.proto.messages.AbstractMessage.QOSType;
import io.netty.channel.Channel;
@@ -43,7 +43,6 @@ import io.netty.channel.ChannelPipeline;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sun.misc.MessageUtils;
/**
* Class responsible to handle the logic of MQTT protocol it's the director of
@@ -406,7 +405,7 @@ public class ProtocolProcessor {
* */
private void republishStoredInSession(ClientSession clientSession) {
LOG.trace("republishStoredInSession for client <{}>", clientSession);
List<IMessagesStore.StoredMessage> publishedEvents = clientSession.storedMessages();
BlockingQueue<StoredMessage> publishedEvents = clientSession.queue();
if (publishedEvents.isEmpty()) {
LOG.info("No stored messages for client <{}>", clientSession.clientID);
return;
@@ -414,9 +413,6 @@ public class ProtocolProcessor {
LOG.info("republishing stored messages to client <{}>", clientSession.clientID);
this.internalRepublisher.publishStored(clientSession, publishedEvents);
for (IMessagesStore.StoredMessage pubEvt : publishedEvents) {
clientSession.removeEnqueued(pubEvt.getGuid());
}
}
public void processPubAck(Channel channel, PubAckMessage msg) {
@@ -438,7 +434,7 @@ public class ProtocolProcessor {
m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username));
}
static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) {
public static IMessagesStore.StoredMessage asStoredMessage(PublishMessage msg) {
IMessagesStore.StoredMessage stored = new IMessagesStore.StoredMessage(msg.getPayload().array(), msg.getQos(), msg.getTopicName());
stored.setRetained(msg.isRetainFlag());
stored.setMessageID(msg.getMessageID());
@@ -641,10 +637,7 @@ public class ProtocolProcessor {
if (descriptor.cleanSession) {
LOG.debug("Removing messages in session's queue for client <{}>", clientID);
Collection<MessageGUID> queue = this.m_sessionsStore.enqueued(clientID);
for (MessageGUID guid : queue) {
this.m_sessionsStore.removeEnqueued(clientID, guid);
}
this.m_sessionsStore.dropQueue(clientID);
LOG.debug("Removed messages in session for client's queue <{}>", clientID);
}
return true;
@@ -848,11 +841,14 @@ public class ProtocolProcessor {
ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
boolean emptyQueue = false;
while (channel.isWritable() && !emptyQueue) {
AbstractMessage msg = clientSession.dequeue();
StoredMessage msg = clientSession.queue().poll();
if (msg == null) {
emptyQueue = true;
} else {
channel.write(msg);
//recreate a publish from stored publish in queue
boolean retained = m_messagesStore.getMessageByGuid(msg.getGuid()) != null;
PublishMessage pubMsg = createPublishForQos(msg.getTopic(), msg.getQos(), msg.getMessage(), retained);
channel.write(pubMsg);
}
}
channel.flush();
broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+1 -10
View File
@@ -78,7 +78,7 @@ class MapDBMessagesStore implements IMessagesStore {
LOG.debug("storePublishForFuture store evt {}", storedMessage);
if (storedMessage.getClientID() == null) {
LOG.error("persisting a message without a clientID, bad programming error msg: {}", storedMessage);
throw new IllegalArgumentException("\"persisting a message without a clientID, bad programming error");
throw new IllegalArgumentException("persisting a message without a clientID, bad programming error");
}
MessageGUID guid = new MessageGUID(UUID.randomUUID().toString());
storedMessage.setGuid(guid);
@@ -89,15 +89,6 @@ class MapDBMessagesStore implements IMessagesStore {
return guid;
}
@Override
public List<StoredMessage> listMessagesInSession(Collection<MessageGUID> guids) {
List<StoredMessage> ret = new ArrayList<>();
for (MessageGUID guid : guids) {
ret.add(m_persistentMessageStore.get(guid));
}
return ret;
}
@Override
public void dropMessagesInSession(String clientID) {
ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java
+6 -17
View File
@@ -16,6 +16,7 @@
package io.moquette.spi.persistence;
import io.moquette.spi.ClientSession;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.IMessagesStore.StoredMessage;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.MessageGUID;
@@ -27,6 +28,7 @@ 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;
@@ -44,8 +46,6 @@ class MapDBSessionsStore implements ISessionsStore {
//map clientID <-> set of currently in flight packet identifiers
private Map<String, Set<Integer>> m_inFlightIds;
private ConcurrentMap<String, PersistentSession> m_persistentSessions;
//maps clientID->[guid*], insertion order cares, it's queue
private ConcurrentMap<String, List<MessageGUID>> m_enqueuedStore;
//maps clientID->[MessageId -> guid]
private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;
@@ -62,7 +62,6 @@ class MapDBSessionsStore implements ISessionsStore {
m_inflightStore = m_db.getHashMap("inflight");
m_inFlightIds = m_db.getHashMap("inflightPacketIDs");
m_persistentSessions = m_db.getHashMap("sessions");
m_enqueuedStore = m_db.getHashMap("sessionQueue");
m_secondPhaseStore = m_db.getHashMap("secondPhase");
}
@@ -210,23 +209,13 @@ class MapDBSessionsStore implements ISessionsStore {
}
@Override
public void bindToDeliver(MessageGUID guid, String clientID) {
List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
guids.add(guid);
m_enqueuedStore.put(clientID, guids);
m_messagesStore.incUsageCounter(guid);
public BlockingQueue<StoredMessage> queue(String clientID) {
return this.m_db.getQueue(clientID);
}
@Override
public Collection<MessageGUID> enqueued(String clientID) {
return Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
}
@Override
public void removeEnqueued(String clientID, MessageGUID guid) {
List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
guids.remove(guid);
m_enqueuedStore.put(clientID, guids);
public void dropQueue(String clientID) {
this.m_db.delete(clientID);
}
@Override