diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java index 64ec8dd8..3adf6959 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java @@ -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 diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java index 047d8e21..eb6a827d 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java @@ -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); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java index 508a929e..59a4c1d6 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java @@ -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: + * + * * @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 storedMessages() { - return messagesStore.listMessagesInSession(clientID); + //read all messages from enqueued store + Collection 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); + } } \ No newline at end of file diff --git a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java index abf0a561..ebd7c0ab 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java @@ -68,22 +68,26 @@ public interface IMessagesStore { */ Collection 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 listMessagesInSession(String clientID); + List listMessagesInSession(Collection 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. diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java index 0369ed70..05959d4b 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java @@ -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 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); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 96e2a009..118dead4 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -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 diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java index 48007897..775b18f1 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java @@ -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{" + diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index f5ff72ef..46feca80 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -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 m_retainedStore; - //maps clientID to the list of pending messages stored - private ConcurrentMap> m_persistentMessageStore; - //bind clientID+MsgID -> evt message published - private ConcurrentMap m_inflightStore; + //maps guid to message, it's message store + private ConcurrentMap m_persistentMessageStore; + //maps clientID->[MessageId -> guid] + private ConcurrentMap> m_inflightStore; //map clientID <-> set of currently in flight packet identifiers Map> 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> m_persistentSubscriptions; private ConcurrentMap m_persistentSessions; + + //maps clientID->[guid*] + private ConcurrentMap> m_enqueuedStore; + //maps clientID->[messageID*] + private ConcurrentMap> 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 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 listMessagesInSession(String clientID) { - List liveEvts = new ArrayList<>(); - List storedEvts = defaultGet(m_persistentMessageStore, clientID, Collections.emptyList()); - - for (StoredPublishEvent storedEvt : storedEvts) { - liveEvts.add(convertFromStored(storedEvt)); + public List listMessagesInSession(Collection guids) { + List 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 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 listMessagesInSession(String clientID) { +// List liveEvts = new ArrayList<>(); +// List storedEvts = defaultGet(m_persistentMessageStore, clientID, Collections.emptyList()); +// +// for (StoredPublishEvent storedEvt : storedEvts) { +// liveEvts.add(convertFromStored(storedEvt)); +// } +// return liveEvts; +// } +// +// @Override +// public void removeMessage(String clientID, Integer messageID) { +// List 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 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 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 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 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 guids = defaultGet(m_enqueuedStore, clientID, new HashSet()); + guids.add(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public Collection enqueued(String clientID) { + return defaultGet(m_enqueuedStore, clientID, new HashSet()); + } + + @Override + public void removeEnqueued(String clientID, String guid) { + Set guids = defaultGet(m_enqueuedStore, clientID, new HashSet()); + guids.remove(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public void secondPhaseAcknowledged(String clientID, int messageID) { + Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.remove(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public void secondPhaseAckWaiting(String clientID, int messageID) { + Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.add(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + private void activationHelper(String clientID, boolean activation) { PersistentSession storedSession = m_persistentSessions.get(clientID); if (storedSession == null) { diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java index 7b179b9e..f670d683 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerIntegrationFuseTest.java @@ -156,8 +156,8 @@ public class ServerIntegrationFuseTest { //reconnect and expect to receive the hello 2 message m_subscriber = m_mqtt.blockingConnection(); m_subscriber.connect(); - topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; - m_subscriber.subscribe(topics); + //topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; + //m_subscriber.subscribe(topics); msg = m_subscriber.receive(); msg.ack(); assertEquals("Hello world MQTT!!-2", new String(msg.getPayload())); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index 28711f67..b606d03f 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -38,8 +38,13 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { private Map m_retainedStore = new HashMap<>(); //TODO move in a multimap because only Qos1 and QoS2 are stored here and they have messageID(key of secondary map) - private Map> m_persistentMessageStore = new HashMap<>(); - private Map m_inflightStore = new HashMap<>(); + private Map m_persistentMessageStore = new HashMap<>(); + //maps clientID->[MessageId -> guid] + private Map> m_inflightStore = new HashMap<>(); + //maps clientID->[guid*] + private Map> m_enqueuedStore = new HashMap<>(); + //maps clientID->[messageID*] + private Map> m_secondPhaseStore = new HashMap<>(); private Map> m_inflightIDs = new HashMap<>(); private Map m_qos2Store = new HashMap<>(); private MemorySessionStore m_sessionsStore; @@ -85,37 +90,21 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { } @Override - public void storePublishForFuture(PublishEvent evt) { + public String storePublishForFuture(PublishEvent evt) { LOG.debug("storePublishForFuture store evt {}", evt); - String clientID = evt.getClientID(); - List storedEvents = defaultGet(m_persistentMessageStore, clientID, new ArrayList()); - storedEvents.add(evt); - m_persistentMessageStore.put(clientID, storedEvents); + String guid = UUID.randomUUID().toString(); + evt.setGuid(guid); + m_persistentMessageStore.put(guid, evt); + return guid; } @Override - public List listMessagesInSession(String clientID) { - return new ArrayList<>(defaultGet(m_persistentMessageStore, clientID, Collections.emptyList())); - } - - @Override - public void removeMessage(String clientID, Integer messageID) { - List events = m_persistentMessageStore.get(clientID); - if (events == null) { - return; + public List listMessagesInSession(Collection guids) { + List ret = new ArrayList<>(); + for (String guid : guids) { + ret.add(m_persistentMessageStore.get(guid)); } - PublishEvent toRemoveEvt = null; - for (PublishEvent evt : events) { - if (evt.getMessageID() == null && messageID == null) { - //was a qos0 message (no ID) - toRemoveEvt = evt; - } - if (evt.getMessageID() == messageID) { - toRemoveEvt = evt; - } - } - events.remove(toRemoveEvt); - m_persistentMessageStore.put(clientID, events); + return ret; } @Override @@ -123,22 +112,6 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { m_persistentMessageStore.remove(clientID); } - @Override - public void cleanTemporaryPublish(String clientID, int packetID) { - String publishKey = String.format("%s%d", clientID, packetID); - m_inflightStore.remove(publishKey); - Set inFlightForClient = 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); - m_inflightStore.put(publishKey, evt); - } - /** * Return the next valid packetIdentifier for the given client session. * */ @@ -229,4 +202,57 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { public void deactivate(String clientID) { m_sessionsStore.deactivate(clientID); } + + @Override + public void inFlightAck(String clientID, int messageID) { + Map 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 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 guids = defaultGet(m_enqueuedStore, clientID, new HashSet()); + guids.add(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public Collection enqueued(String clientID) { + return defaultGet(m_enqueuedStore, clientID, new HashSet()); + } + + @Override + public void removeEnqueued(String clientID, String guid) { + Set guids = defaultGet(m_enqueuedStore, clientID, new HashSet()); + guids.remove(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public void secondPhaseAcknowledged(String clientID, int messageID) { + Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.remove(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public void secondPhaseAckWaiting(String clientID, int messageID) { + Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.add(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 970d7cdc..193ae611 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -368,7 +368,8 @@ public class ProtocolProcessorTest { m_processor.processConnect(m_session, connectMessage); //Verify no messages are still stored - assertTrue(m_storageService.listMessagesInSession(FAKE_PUBLISHER_ID).isEmpty()); + Collection guids = m_sessionStore.enqueued(FAKE_PUBLISHER_ID); + assertTrue(m_storageService.listMessagesInSession(guids).isEmpty()); } @Test @@ -483,8 +484,8 @@ public class ProtocolProcessorTest { ProtocolProcessor processor = new ProtocolProcessor() { @Override - protected void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, - boolean retained, Integer messageID) { + protected void directSend(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, + boolean retained, Integer messageID) { publishedForwarded.add(new PublishEvent(topic, qos, message, retained, clientId, messageID)); } }; diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index e0b5a47d..39c323ea 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -243,7 +243,7 @@ public class ProtocolProcessor_CONNECT_Test { //Connect a first time m_processor.processConnect(m_session, connMsg); //disconnect - m_processor.processDisconnect(m_session, new DisconnectMessage()); + m_processor.processDisconnect(m_session); //Exercise, reconnect MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); @@ -278,7 +278,7 @@ public class ProtocolProcessor_CONNECT_Test { assertTrue(subscriptions.contains(expectedSubscription)); //disconnect - m_processor.processDisconnect(m_session, new DisconnectMessage()); + m_processor.processDisconnect(m_session); //reconnect clean session a false m_processor.processConnect(m_session, connMsg); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java index 5d970ba0..189518f9 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -91,7 +91,7 @@ public class MapDBPersistentStoreTest { assertEquals(1, packetId); //release the ID - m_storageService.cleanTemporaryPublish("CLIENT", packetId); +// m_storageService.cleanTemporaryPublish("CLIENT", packetId); //request a second packetID, counter restarts from 0 packetId = m_storageService.nextPacketID("CLIENT"); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java index cd8db2f6..e77adc72 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MemorySessionStore.java @@ -19,6 +19,8 @@ import org.eclipse.moquette.spi.ClientSession; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; + +import static org.eclipse.moquette.spi.impl.Utils.defaultGet; import static org.eclipse.moquette.spi.persistence.MapDBPersistentStore.PersistentSession; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,6 +38,13 @@ public class MemorySessionStore implements ISessionsStore { private Map m_persistentSessions = new HashMap<>(); + //maps clientID->[MessageId -> guid] + private Map> m_inflightStore = new HashMap<>(); + //maps clientID->[guid*] + private Map> m_enqueuedStore = new HashMap<>(); + //maps clientID->[messageID*] + private Map> m_secondPhaseStore = new HashMap<>(); + private final IMessagesStore m_messagesStore; public MemorySessionStore(IMessagesStore messagesStore) { @@ -146,4 +155,57 @@ public class MemorySessionStore implements ISessionsStore { PersistentSession newStoredSession = new PersistentSession(storedSession.cleanSession, activation); m_persistentSessions.put(clientID, newStoredSession); } + + @Override + public void inFlightAck(String clientID, int messageID) { + Map 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 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 guids = defaultGet(m_enqueuedStore, clientID, new HashSet()); + guids.add(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public Collection enqueued(String clientID) { + return defaultGet(m_enqueuedStore, clientID, new HashSet()); + } + + @Override + public void removeEnqueued(String clientID, String guid) { + Set guids = defaultGet(m_enqueuedStore, clientID, new HashSet()); + guids.remove(guid); + m_enqueuedStore.put(clientID, guids); + } + + @Override + public void secondPhaseAcknowledged(String clientID, int messageID) { + Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.remove(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } + + @Override + public void secondPhaseAckWaiting(String clientID, int messageID) { + Set messageIDs = defaultGet(m_secondPhaseStore, clientID, new HashSet()); + messageIDs.add(messageID); + m_secondPhaseStore.put(clientID, messageIDs); + } }