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

First refactoring using the concept of ClientSession also for publish

Browse Source
This commit is contained in:
andsel
2015-11-15 11:16:18 +01:00
parent ce1618cb52
commit 191a94e19d
14 changed files with 415 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java
+11 -21
View File
@@ -17,13 +17,7 @@ package org.eclipse.moquette.server.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
@@ -36,30 +30,26 @@ import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import java.io.*;
import java.net.URL;
import java.security.*;
import java.security.cert.CertificateException;
import java.util.List;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import io.netty.util.concurrent.Future;
import org.eclipse.moquette.commons.Constants;
import org.eclipse.moquette.server.config.IConfig;
import org.eclipse.moquette.spi.IMessaging;
import org.eclipse.moquette.parser.netty.MQTTDecoder;
import org.eclipse.moquette.parser.netty.MQTTEncoder;
import org.eclipse.moquette.server.ServerAcceptor;
import org.eclipse.moquette.server.config.IConfig;
import org.eclipse.moquette.server.netty.metrics.*;
import org.eclipse.moquette.spi.impl.ProtocolProcessor;
import org.eclipse.moquette.spi.impl.SimpleMessaging;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import java.io.*;
import java.net.URL;
import java.security.*;
import java.security.cert.CertificateException;
import java.util.List;
/**
*
* @author andrea
broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java
+1 -1
View File
@@ -79,7 +79,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
m_processor.processPubRel(new NettyChannel(ctx), (PubRelMessage) msg);
break;
case DISCONNECT:
m_processor.processDisconnect(new NettyChannel(ctx), (DisconnectMessage) msg);
m_processor.processDisconnect(new NettyChannel(ctx));
break;
case PUBACK:
m_processor.processPubAck(new NettyChannel(ctx), (PubAckMessage) msg);
broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java
+46 -8
View File
@@ -22,11 +22,25 @@ import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
/**
* Model a Session like describe on page 25 of MQTT 3.1.1 specification:
* The Session state in the Server consists of:
* <ul>
* <li>The existence of a Session, even if the rest of the Session state is empty.</li>
* <li>The Client’s subscriptions.</li>
* <li>QoS 1 and QoS 2 messages which have been sent to the Client, but have not been
* completely acknowledged.</li>
* <li>QoS 1 and QoS 2 messages pending transmission to the Client.</li>
* <li>QoS 2 messages which have been received from the Client, but have not been
* completely acknowledged.</li>
* <li>Optionally, QoS 0 messages pending transmission to the Client.</li>
* </ul>
*
* @author andrea
*/
public class ClientSession {
@@ -60,21 +74,21 @@ public class ClientSession {
* @return the list of messages to be delivered for client related to the session.
* */
public List<PublishEvent> storedMessages() {
return messagesStore.listMessagesInSession(clientID);
//read all messages from enqueued store
Collection<String> guids = this.m_sessionsStore.enqueued(clientID);
return messagesStore.listMessagesInSession(guids);
}
/**
* Remove a message previously stored for delivery.
* */
public void removeDelivered(int messageID) {
messagesStore.removeMessage(clientID, messageID);
* Remove the messages stored in a cleanSession false.
*/
public void removeEnqueued(String guid) {
this.m_sessionsStore.removeEnqueued(this.clientID, guid);
}
@Override
public String toString() {
return "ClientSession{" +
"clientID='" + clientID + '\'' +
'}';
return "ClientSession{clientID='" + clientID + '\'' +"}";
}
public boolean subscribe(String topicFilter, Subscription newSubscription) {
@@ -133,4 +147,28 @@ public class ClientSession {
public boolean isActive() {
return this.active;
}
public int nextPacketId() {
return this.messagesStore.nextPacketID(this.clientID);
}
public void inFlightAcknowledged(int messageID) {
m_sessionsStore.inFlightAck(this.clientID, messageID);
}
public void inFlightAckWaiting(String guid, int messageID) {
m_sessionsStore.inFlight(this.clientID, messageID, guid);
}
public void secondPhaseAcknowledged(int messageID) {
m_sessionsStore.secondPhaseAcknowledged(clientID, messageID);
}
public void secondPhaseAckWaiting(int messageID) {
m_sessionsStore.secondPhaseAckWaiting(clientID, messageID);
}
public void enqueueToDeliver(String guid) {
this.m_sessionsStore.bindToDeliver(guid, this.clientID);
}
}
broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java
+9 -5
View File
@@ -68,22 +68,26 @@ public interface IMessagesStore {
*/
Collection<StoredMessage> searchMatching(IMatchingCondition condition);
void storePublishForFuture(PublishEvent evt);
/**
* Persist the message.
* @return the unique id in the storage (guid).
* */
String storePublishForFuture(PublishEvent evt);
/**
* 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.
*/
List<PublishEvent> listMessagesInSession(String clientID);
List<PublishEvent> listMessagesInSession(Collection<String> guids);
void removeMessage(String clientID, Integer packetID);
// void removeMessage(String clientID, Integer packetID);
void dropMessagesInSession(String clientID);
void cleanTemporaryPublish(String clientID, int packetID);
// void cleanTemporaryPublish(String clientID, int packetID);
void storeTemporaryPublish(PublishEvent evt, String clientID, int packetID);
// void storeTemporaryPublish(PublishEvent evt, String clientID, int packetID);
/**
* Return the next valid packetIdentifer for the given client session.
broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java
+27
View File
@@ -15,6 +15,7 @@
*/
package org.eclipse.moquette.spi;
import java.util.Collection;
import java.util.List;
import java.util.Set;
@@ -64,4 +65,30 @@ public interface ISessionsStore {
void activate(String clientID);
void deactivate(String clientID);
void inFlightAck(String clientID, int messageID);
/**
* Save the binding messageID, clientID <-> guid
* */
void inFlight(String clientID, int messageID, String guid);
/**
* Store the guid to be later published.
* */
void bindToDeliver(String guid, String clientID);
/**
* List the guids for retained messages for the session
* */
Collection<String> enqueued(String clientID);
/**
* Remove form the queue of stored messages for session.
* */
void removeEnqueued(String clientID, String guid);
void secondPhaseAcknowledged(String clientID, int messageID);
void secondPhaseAckWaiting(String clientID, int messageID);
}
broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+42 -25
View File
@@ -34,7 +34,6 @@ import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.proto.messages.AbstractMessage.QOSType;
import org.eclipse.moquette.proto.messages.ConnAckMessage;
import org.eclipse.moquette.proto.messages.ConnectMessage;
import org.eclipse.moquette.proto.messages.DisconnectMessage;
import org.eclipse.moquette.proto.messages.PubAckMessage;
import org.eclipse.moquette.proto.messages.PubCompMessage;
import org.eclipse.moquette.proto.messages.PubRecMessage;
@@ -231,6 +230,7 @@ public class ProtocolProcessor {
//force the republish of stored QoS1 and QoS2
republishStoredInSession(clientSession);
}
LOG.info("CONNECT processed");
}
private void failedCredentials(ServerChannel session) {
@@ -253,9 +253,10 @@ public class ProtocolProcessor {
LOG.info("republishing stored messages to client <{}>", clientSession.clientID);
for (PublishEvent pubEvt : publishedEvents) {
sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID());
clientSession.removeDelivered(pubEvt.getMessageID());
//TODO put in flight zone
directSend(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
pubEvt.getMessage(), false, pubEvt.getMessageID());
clientSession.removeEnqueued(pubEvt.getGuid());
}
}
@@ -263,7 +264,8 @@ public class ProtocolProcessor {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
int messageID = msg.getMessageID();
//Remove the message from message store
m_messagesStore.removeMessage(clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
targetSession.inFlightAcknowledged(messageID);
}
public void processPublish(ServerChannel session, PublishMessage msg) {
@@ -292,10 +294,11 @@ public class ProtocolProcessor {
if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0
forward2Subscribers(publishEvt);
} else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1
//TODO implement inFlight!!
m_messagesStore.storeTemporaryPublish(publishEvt, clientID, messageID);
//TODO implement inFlight!!boolean retain = pubEvt.isRetain();
// final Integer messageID = pubEvt.getMessageID();
// m_messagesStore.storeTemporaryPublish(publishEvt, clientID, messageID);
forward2Subscribers(publishEvt);
m_messagesStore.cleanTemporaryPublish(clientID, messageID);
// m_messagesStore.cleanTemporaryPublish(clientID, messageID);
//NB the PUB_ACK could be sent also after the addInFlight
sendPubAck(new PubAckEvent(messageID, clientID));
LOG.debug("replying with PubAck to MSG ID {}", messageID);
@@ -338,17 +341,24 @@ public class ProtocolProcessor {
/**
* Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
* */
//TODO rename to route2Subscribers
void forward2Subscribers(PublishEvent pubEvt) {
final String topic = pubEvt.getTopic();
final AbstractMessage.QOSType publishingQos = pubEvt.getQos();
final ByteBuffer origMessage = pubEvt.getMessage();
boolean retain = pubEvt.isRetain();
final Integer messageID = pubEvt.getMessageID();
// boolean retain = pubEvt.isRetain();
// final Integer messageID = pubEvt.getMessageID();
LOG.debug("forward2Subscribers republishing to existing subscribers that matches the topic {}", topic);
if (LOG.isDebugEnabled()) {
LOG.debug("content <{}>", DebugUtils.payload2Str(origMessage));
LOG.debug("subscription tree {}", subscriptions.dumpTree());
}
//if QoS 1 or 2 store the message
String guid = null;
if (publishingQos == QOSType.EXACTLY_ONCE || publishingQos == QOSType.LEAST_ONE) {
guid = m_messagesStore.storePublishForFuture(pubEvt);
}
for (final Subscription sub : subscriptions.matches(topic)) {
AbstractMessage.QOSType qos = publishingQos;
if (qos.byteValue() > sub.getRequestedQos().byteValue()) {
@@ -361,33 +371,34 @@ public class ProtocolProcessor {
ByteBuffer message = origMessage.duplicate();
if (qos == AbstractMessage.QOSType.MOST_ONE && targetSession.isActive()) {
//QoS 0
sendPublish(sub.getClientId(), topic, qos, message, false, null);
directSend(sub.getClientId(), topic, qos, message, false, null);
} else {
//QoS 1 or 2
//if the target subscription is not clean session and is not connected => store it
if (!sub.isCleanSession() && !targetSession.isActive()) {
//clone the event with matching clientID
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0);
m_messagesStore.storePublishForFuture(newPublishEvt);
//store the message in targetSession queue to deliver
targetSession.enqueueToDeliver(guid);
} else {
//TODO also QoS 1 has to be stored in Flight Zone
//if QoS 2 then store it in temp memory
if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0);
m_messagesStore.storeTemporaryPublish(newPublishEvt, sub.getClientId(), messageID);
}
// if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) {
// PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), messageID != null ? messageID : 0);
// m_messagesStore.storeTemporaryPublish(newPublishEvt, sub.getClientId(), messageID);
// }
//publish
if (targetSession.isActive()) {
int messageId = m_messagesStore.nextPacketID(sub.getClientId());
sendPublish(sub.getClientId(), topic, qos, message, false, messageId);
// int messageId = m_messagesStore.nextPacketID(sub.getClientId());
int messageId = targetSession.nextPacketId();
targetSession.inFlightAckWaiting(guid, messageId);
directSend(sub.getClientId(), topic, qos, message, false, messageId);
}
}
}
}
}
protected void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) {
LOG.debug("sendPublish invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID);
protected void directSend(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, Integer messageID) {
LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}", clientId, topic, qos, retained, messageID);
PublishMessage pubMessage = new PublishMessage();
pubMessage.setRetainFlag(retained);
pubMessage.setTopicName(topic);
@@ -487,6 +498,10 @@ public class ProtocolProcessor {
public void processPubRec(ServerChannel session, PubRecMessage msg) {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
int messageID = msg.getMessageID();
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
//remove from the inflight and move to the QoS2 second phase queue
targetSession.inFlightAcknowledged(messageID);
targetSession.secondPhaseAckWaiting(messageID);
//once received a PUBREC reply with a PUBREL(messageID)
LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientID {} ad messageID {}", clientID, messageID);
PubRelMessage pubRelMessage = new PubRelMessage();
@@ -501,10 +516,12 @@ public class ProtocolProcessor {
int messageID = msg.getMessageID();
LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
//once received the PUBCOMP then remove the message from the temp memory
m_messagesStore.cleanTemporaryPublish(clientID, messageID);
// m_messagesStore.cleanTemporaryPublish(clientID, messageID);
ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
targetSession.secondPhaseAcknowledged(messageID);
}
public void processDisconnect(ServerChannel session, DisconnectMessage msg) throws InterruptedException {
public void processDisconnect(ServerChannel session) throws InterruptedException {
String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
@@ -620,7 +637,7 @@ public class ProtocolProcessor {
//forwardPublishQoS0(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null :
m_messagesStore.nextPacketID(newSubscription.getClientId());
sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
directSend(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
}
//notify the Observables
broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java
+9
View File
@@ -33,6 +33,7 @@ public class PublishEvent extends MessagingEvent {
String m_clientID;
//Optional attribute, available only fo QoS 1 and 2
Integer m_msgID;
private String guid;
public PublishEvent(String topic, QOSType qos, ByteBuffer message, boolean retain,
String clientID, Integer msgID) {
@@ -81,6 +82,14 @@ public class PublishEvent extends MessagingEvent {
return m_msgID;
}
public void setGuid(String guid) {
this.guid = guid;
}
public String getGuid() {
return guid;
}
@Override
public String toString() {
return "PublishEvent{" +
broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
+129 -63
View File
@@ -17,6 +17,7 @@
package org.eclipse.moquette.spi.persistence;
import org.eclipse.moquette.proto.MQTTException;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import org.eclipse.moquette.spi.ClientSession;
import org.eclipse.moquette.spi.IMatchingCondition;
import org.eclipse.moquette.spi.IMessagesStore;
@@ -24,10 +25,6 @@ import org.eclipse.moquette.spi.ISessionsStore;
import org.eclipse.moquette.spi.impl.events.PublishEvent;
import org.eclipse.moquette.spi.impl.storage.StoredPublishEvent;
import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
import org.eclipse.moquette.proto.messages.AbstractMessage;
import static org.eclipse.moquette.spi.impl.Utils.defaultGet;
import org.mapdb.DB;
import org.mapdb.DBMaker;
import org.slf4j.Logger;
@@ -38,7 +35,12 @@ import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import static org.eclipse.moquette.spi.impl.Utils.defaultGet;
/**
* MapDB main persistence implementation
@@ -62,10 +64,10 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class);
private ConcurrentMap<String, StoredMessage> m_retainedStore;
//maps clientID to the list of pending messages stored
private ConcurrentMap<String, List<StoredPublishEvent>> m_persistentMessageStore;
//bind clientID+MsgID -> evt message published
private ConcurrentMap<String, StoredPublishEvent> m_inflightStore;
//maps guid to message, it's message store
private ConcurrentMap<String, StoredPublishEvent> m_persistentMessageStore;
//maps clientID->[MessageId -> guid]
private ConcurrentMap<String, Map<Integer, String>> m_inflightStore;
//map clientID <-> set of currently in flight packet identifiers
Map<String, Set<Integer>> m_inFlightIds;
//bind clientID+MsgID -> evt message published
@@ -73,6 +75,12 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
//persistent Map of clientID, set of Subscriptions
private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
private ConcurrentMap<String, PersistentSession> m_persistentSessions;
//maps clientID->[guid*]
private ConcurrentMap<String, Set<String>> m_enqueuedStore;
//maps clientID->[messageID*]
private ConcurrentMap<String, Set<Integer>> m_secondPhaseStore;
private DB m_db;
private String m_storePath;
@@ -113,6 +121,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
m_persistentSubscriptions = m_db.getHashMap("subscriptions");
m_qos2Store = m_db.getHashMap("qos2Store");
m_persistentSessions = m_db.getHashMap("sessions");
m_enqueuedStore = m_db.getHashMap("sessionQueue");
m_secondPhaseStore = m_db.getHashMap("secondPhase");
m_scheduler.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
@@ -155,72 +165,75 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
}
@Override
public void storePublishForFuture(PublishEvent evt) {
List<StoredPublishEvent> storedEvents;
String clientID = evt.getClientID();
if (!m_persistentMessageStore.containsKey(clientID)) {
storedEvents = new ArrayList<>();
} else {
storedEvents = m_persistentMessageStore.get(clientID);
}
storedEvents.add(convertToStored(evt));
m_persistentMessageStore.put(clientID, storedEvents);
//NB rewind the evt message content
LOG.debug("Stored published message for client <{}> on topic <{}>", clientID, evt.getTopic());
public String storePublishForFuture(PublishEvent evt) {
LOG.debug("storePublishForFuture store evt {}", evt);
String guid = UUID.randomUUID().toString();
evt.setGuid(guid);
m_persistentMessageStore.put(guid, convertToStored(evt));
return guid;
}
@Override
public List<PublishEvent> listMessagesInSession(String clientID) {
List<PublishEvent> liveEvts = new ArrayList<>();
List<StoredPublishEvent> storedEvts = defaultGet(m_persistentMessageStore, clientID, Collections.<StoredPublishEvent>emptyList());
for (StoredPublishEvent storedEvt : storedEvts) {
liveEvts.add(convertFromStored(storedEvt));
public List<PublishEvent> listMessagesInSession(Collection<String> guids) {
List<PublishEvent> ret = new ArrayList<>();
for (String guid : guids) {
ret.add(convertFromStored(m_persistentMessageStore.get(guid))) ;
}
return liveEvts;
return ret;
}
@Override
public void removeMessage(String clientID, Integer messageID) {
List<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
if (events == null) {
return;
}
StoredPublishEvent toRemoveEvt = null;
for (StoredPublishEvent evt : events) {
if (evt.getMessageID() == null && messageID == null) {
//was a qos0 message (no ID)
toRemoveEvt = evt;
}
if (evt.getMessageID().equals(messageID)) {
toRemoveEvt = evt;
}
}
events.remove(toRemoveEvt);
m_persistentMessageStore.put(clientID, events);
}
// @Override
// public List<PublishEvent> listMessagesInSession(String clientID) {
// List<PublishEvent> liveEvts = new ArrayList<>();
// List<StoredPublishEvent> storedEvts = defaultGet(m_persistentMessageStore, clientID, Collections.<StoredPublishEvent>emptyList());
//
// for (StoredPublishEvent storedEvt : storedEvts) {
// liveEvts.add(convertFromStored(storedEvt));
// }
// return liveEvts;
// }
//
// @Override
// public void removeMessage(String clientID, Integer messageID) {
// List<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
// if (events == null) {
// return;
// }
// StoredPublishEvent toRemoveEvt = null;
// for (StoredPublishEvent evt : events) {
// if (evt.getMessageID() == null && messageID == null) {
// //was a qos0 message (no ID)
// toRemoveEvt = evt;
// }
// if (evt.getMessageID().equals(messageID)) {
// toRemoveEvt = evt;
// }
// }
// events.remove(toRemoveEvt);
// m_persistentMessageStore.put(clientID, events);
// }
public void dropMessagesInSession(String clientID) {
m_persistentMessageStore.remove(clientID);
}
//----------------- In flight methods -----------------
@Override
public void cleanTemporaryPublish(String clientID, int packetID) {
String publishKey = String.format("%s%d", clientID, packetID);
m_inflightStore.remove(publishKey);
Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
if (inFlightForClient != null) {
inFlightForClient.remove(packetID);
}
}
@Override
public void storeTemporaryPublish(PublishEvent evt, String clientID, int packetID) {
String publishKey = String.format("%s%d", clientID, packetID);
StoredPublishEvent storedEvt = convertToStored(evt);
m_inflightStore.put(publishKey, storedEvt);
}
// @Override
// public void cleanTemporaryPublish(String clientID, int packetID) {
// String publishKey = String.format("%s%d", clientID, packetID);
// m_inflightStore.remove(publishKey);
// Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
// if (inFlightForClient != null) {
// inFlightForClient.remove(packetID);
// }
// }
//
// @Override
// public void storeTemporaryPublish(PublishEvent evt, String clientID, int packetID) {
// String publishKey = String.format("%s%d", clientID, packetID);
// StoredPublishEvent storedEvt = convertToStored(evt);
// m_inflightStore.put(publishKey, storedEvt);
// }
/**
* Return the next valid packetIdentifier for the given client session.
@@ -356,6 +369,59 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
activationHelper(clientID, false);
}
@Override
public void inFlightAck(String clientID, int messageID) {
Map<Integer, String> m = this.m_inflightStore.get(clientID);
if (m == null) {
LOG.error("Can't find the inFlight record for client <{}>", clientID);
return;
}
m.remove(messageID);
}
@Override
public void inFlight(String clientID, int messageID, String guid) {
Map<Integer, String> m = this.m_inflightStore.get(clientID);
if (m == null) {
m = new HashMap<>();
}
m.put(messageID, guid);
this.m_inflightStore.put(clientID, m);
}
@Override
public void bindToDeliver(String guid, String clientID) {
Set<String> guids = defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
guids.add(guid);
m_enqueuedStore.put(clientID, guids);
}
@Override
public Collection<String> enqueued(String clientID) {
return defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
}
@Override
public void removeEnqueued(String clientID, String guid) {
Set<String> guids = defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
guids.remove(guid);
m_enqueuedStore.put(clientID, guids);
}
@Override
public void secondPhaseAcknowledged(String clientID, int messageID) {
Set<Integer> messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
messageIDs.remove(messageID);
m_secondPhaseStore.put(clientID, messageIDs);
}
@Override
public void secondPhaseAckWaiting(String clientID, int messageID) {
Set<Integer> messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet<Integer>());
messageIDs.add(messageID);
m_secondPhaseStore.put(clientID, messageIDs);
}
private void activationHelper(String clientID, boolean activation) {
PersistentSession storedSession = m_persistentSessions.get(clientID);
if (storedSession == null) {