diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index d13976a3..992673fa 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -82,6 +82,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { } } catch (Exception ex) { LOG.error("Bad error in processing the message", ex); + ctx.fireExceptionCaught(ex); } } diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 520d1b33..22683b5b 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -72,9 +72,6 @@ public class ClientSession { this.m_sessionsStore = sessionsStore; this.cleanSession = cleanSession; } - //List of client's subscriptions - //list of messages not acknowledged by client - //list of in-flight messages /** * @return the list of messages to be delivered for client related to the session. @@ -174,19 +171,18 @@ public class ClientSession { } public void inFlightAcknowledged(int messageID) { + LOG.trace("Acknowledging inflight, clientID <{}> messageID {}", this.clientID, messageID); m_sessionsStore.inFlightAck(this.clientID, messageID); } public void inFlightAckWaiting(String guid, int messageID) { + LOG.trace("Adding to inflight {}, guid <{}>", messageID, guid); 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 IMessagesStore.StoredMessage secondPhaseAcknowledged(int messageID) { + String guid = m_sessionsStore.secondPhaseAcknowledged(clientID, messageID); + return messagesStore.getMessageByGuid(guid); } public void enqueueToDeliver(String guid) { @@ -209,4 +205,12 @@ public class ClientSession { public AbstractMessage dequeue() { return m_queueToPublish.poll(); } + + public void moveInFlightToSecondPhaseAckWaiting(int messageID) { + m_sessionsStore.moveInFlightToSecondPhaseAckWaiting(this.clientID, messageID); + } + + public IMessagesStore.StoredMessage getInflightMessage(int messageID) { + return m_sessionsStore.getInflightMessage(clientID, messageID); + } } \ No newline at end of file diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 3b9accfb..e1237b01 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -28,8 +28,6 @@ import io.moquette.spi.impl.subscriptions.Subscription; */ public interface ISessionsStore { - void updateCleanStatus(String clientID, boolean cleanSession); - class ClientTopicCouple { public final String topicFilter; public final String clientID; @@ -69,6 +67,8 @@ public interface ISessionsStore { void initStore(); + void updateCleanStatus(String clientID, boolean cleanSession); + /** * Add a new subscription to the session */ @@ -134,11 +134,14 @@ public interface ISessionsStore { * */ void removeEnqueued(String clientID, String guid); - void secondPhaseAcknowledged(String clientID, int messageID); + void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID); - void secondPhaseAckWaiting(String clientID, int messageID); + /** + * @return the guid of message just acked. + * */ + String secondPhaseAcknowledged(String clientID, int messageID); String mapToGuid(String clientID, int messageID); - public StoredMessage getInflightMessage(String clientID, int messageID); + StoredMessage getInflightMessage(String clientID, int messageID); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 7c484f58..77487772 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -295,11 +295,12 @@ public class ProtocolProcessor { String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); String username = NettyUtils.userName(channel); - StoredMessage inflightMsg = m_sessionsStore.getInflightMessage(clientID, messageID); + LOG.trace("retrieving inflight for messageID <{}>", messageID); //Remove the message from message store ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); + StoredMessage inflightMsg = targetSession.getInflightMessage(messageID); targetSession.inFlightAcknowledged(messageID); String topic = inflightMsg.getTopic(); @@ -448,7 +449,9 @@ public class ProtocolProcessor { guid = m_messagesStore.storePublishForFuture(pubMsg); } - for (final Subscription sub : subscriptions.matches(topic)) { + List topicMatchingSubscriptions = subscriptions.matches(topic); + LOG.trace("Found {} matching subscriptions to <{}>", topicMatchingSubscriptions.size(), topic); + for (final Subscription sub : topicMatchingSubscriptions) { AbstractMessage.QOSType qos = publishingQos; if (qos.byteValue() > sub.getRequestedQos().byteValue()) { qos = sub.getRequestedQos(); @@ -509,7 +512,6 @@ public class ProtocolProcessor { throw new RuntimeException("Internal bad error, found m_clientIDs to null while it should be " + "initialized, somewhere it's overwritten!!"); } - //LOG.trace("clientIDs are {}", m_clientIDs); if (m_clientIDs.get(clientId) == null) { //TODO while we were publishing to the target client, that client disconnected, // could happen is not an error HANDLE IT @@ -588,12 +590,11 @@ public class ProtocolProcessor { public void processPubRec(Channel channel, PubRecMessage msg) { String clientID = NettyUtils.clientID(channel); - int messageID = msg.getMessageID(); ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); //remove from the inflight and move to the QoS2 second phase queue - targetSession.inFlightAcknowledged(messageID); - targetSession.secondPhaseAckWaiting(messageID); + int messageID = msg.getMessageID(); + targetSession.moveInFlightToSecondPhaseAckWaiting(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(); @@ -606,16 +607,14 @@ public class ProtocolProcessor { public void processPubComp(Channel channel, PubCompMessage msg) { String clientID = NettyUtils.clientID(channel); int messageID = msg.getMessageID(); - StoredMessage inflightMsg = m_sessionsStore.getInflightMessage( clientID, messageID ); - 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 ClientSession targetSession = m_sessionsStore.sessionForClient(clientID); verifyToActivate(clientID, targetSession); - targetSession.secondPhaseAcknowledged(messageID); + StoredMessage inflightMsg = targetSession.secondPhaseAcknowledged(messageID); String username = NettyUtils.userName(channel); String topic = inflightMsg.getTopic(); - m_interceptor.notifyMessageAcknowledged( new InterceptAcknowledgedMessage(inflightMsg, topic, username) ); + m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username)); } public void processDisconnect(Channel channel) throws InterruptedException { @@ -725,16 +724,25 @@ public class ProtocolProcessor { if (LOG.isTraceEnabled()) { LOG.trace("subscription tree {}", subscriptions.dumpTree()); } + + for (Subscription subscription : newSubscriptions) { + subscribeSingleTopic(subscription); + } channel.writeAndFlush(ackMessage); - //fire the publish - for(Subscription subscription : newSubscriptions) { - subscribeSingleTopic(subscription, username); + //fire the persisted messages in session + for (Subscription subscription : newSubscriptions) { + publishStoredMessagesInSession(subscription, username); } } - private boolean subscribeSingleTopic(final Subscription newSubscription, String username) { + private void subscribeSingleTopic(final Subscription newSubscription) { + LOG.debug("Subscribing {}", newSubscription); subscriptions.add(newSubscription.asClientTopicCouple()); + } + + private void publishStoredMessagesInSession(final Subscription newSubscription, String username) { + LOG.debug("Publish persisted messages in session {}", newSubscription); //scans retained messages to be published to the new subscription //TODO this is ugly, it does a linear scan on potential big dataset @@ -748,17 +756,18 @@ public class ProtocolProcessor { ClientSession targetSession = m_sessionsStore.sessionForClient(newSubscription.getClientId()); verifyToActivate(newSubscription.getClientId(), targetSession); for (IMessagesStore.StoredMessage storedMsg : messages) { - //fire the as retained the message - LOG.debug("send publish message for topic {}", newSubscription.getTopicFilter()); - //forwardPublishQoS0(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true); - Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null : - targetSession.nextPacketId(); + //fire as retained the message + LOG.trace("send publish message for topic {}", newSubscription.getTopicFilter()); + Integer packetID = storedMsg.getQos() == QOSType.MOST_ONE ? null : targetSession.nextPacketId(); + if (packetID != null) { + LOG.trace("Adding to inflight <{}>", packetID); + targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID); + } directSend(targetSession, storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID); } //notify the Observables m_interceptor.notifyTopicSubscribed(newSubscription, username); - return true; } public void notifyChannelWritable(Channel channel) { diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java index a7d2cf18..716ffcaa 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java @@ -81,6 +81,7 @@ class MapDBMessagesStore implements IMessagesStore { } String guid = UUID.randomUUID().toString(); evt.setGuid(guid); + LOG.debug("storePublishForFuture guid <{}>", guid); m_persistentMessageStore.put(guid, evt); ConcurrentMap messageIdToGuid = m_db.getHashMap(MapDBSessionsStore.messageId2GuidsMapName(evt.getClientID())); messageIdToGuid.put(evt.getMessageID(), guid); diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java index bcfad2d9..0d7ead81 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java @@ -45,8 +45,8 @@ class MapDBSessionsStore implements ISessionsStore { private ConcurrentMap m_persistentSessions; //maps clientID->[guid*], insertion order cares, it's queue private ConcurrentMap> m_enqueuedStore; - //maps clientID->[messageID*] - private ConcurrentMap> m_secondPhaseStore; + //maps clientID->[MessageId -> guid] + private ConcurrentMap> m_secondPhaseStore; private final DB m_db; private final IMessagesStore m_messagesStore; @@ -172,6 +172,7 @@ class MapDBSessionsStore implements ISessionsStore { @Override public void inFlightAck(String clientID, int messageID) { + LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); Map m = this.m_inflightStore.get(clientID); if (m == null) { LOG.error("Can't find the inFlight record for client <{}>", clientID); @@ -193,6 +194,7 @@ class MapDBSessionsStore implements ISessionsStore { m = new HashMap<>(); } m.put(messageID, guid); + LOG.info("storing inflight clientID <{}> messageID {} guid <{}>", clientID, messageID, guid); this.m_inflightStore.put(clientID, m); } @@ -216,17 +218,33 @@ class MapDBSessionsStore implements ISessionsStore { } @Override - public void secondPhaseAcknowledged(String clientID, int messageID) { - Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.remove(messageID); + public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { + LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + LOG.error("Can't find the inFlight record for client <{}>", clientID); + return; + } + String guid = m.remove(messageID); + + //remove from the ids store + Set inFlightForClient = this.m_inFlightIds.get(clientID); + if (inFlightForClient != null) { + inFlightForClient.remove(messageID); + } + + LOG.info("Moving to second phase store"); + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); + messageIDs.put(messageID, guid); m_secondPhaseStore.put(clientID, messageIDs); } @Override - public void secondPhaseAckWaiting(String clientID, int messageID) { - Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.add(messageID); + public String secondPhaseAcknowledged(String clientID, int messageID) { + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); + String guid = messageIDs.remove(messageID); m_secondPhaseStore.put(clientID, messageIDs); + return guid; } @Override @@ -240,13 +258,14 @@ class MapDBSessionsStore implements ISessionsStore { } @Override - public StoredMessage getInflightMessage( String clientID, int messageID ) { - Map clientEntries = m_inflightStore.get( clientID ); - if( clientEntries == null ) + public StoredMessage getInflightMessage(String clientID, int messageID) { + Map clientEntries = m_inflightStore.get(clientID); + if (clientEntries == null) return null; - String guid = clientEntries.get( messageID ); - if( guid == null ) + String guid = clientEntries.get(messageID); + LOG.info("inflight messageID {} guid <{}>", messageID, guid); + if (guid == null) return null; - return m_messagesStore.getMessageByGuid( guid ); + return m_messagesStore.getMessageByGuid(guid); } } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java index 8def3ec8..d17fdfd8 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java @@ -139,8 +139,6 @@ public class ServerIntegrationEmbeddedPublishTest { LOG.info("*** testClientSubscribeBeforeRetainedQoS0IsSent ***"); subscribeToWithQos("/topic", 0); - //super ugly but we need the MQTT client lib finish it's job before us - Thread.sleep(1000); //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, true); @@ -155,6 +153,7 @@ public class ServerIntegrationEmbeddedPublishTest { //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, true); + //LOG.info("** post internalPublish **"); subscribeToWithQos("/topic", 0); //Verify @@ -166,8 +165,6 @@ public class ServerIntegrationEmbeddedPublishTest { LOG.info("*** testClientSubscribeBeforeNotRetainedQoS1IsSent ***"); subscribeToWithQos("/topic", 1); - //super ugly but we need the MQTT client lib finish it's job before us - Thread.sleep(1000); //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, false); @@ -193,8 +190,6 @@ public class ServerIntegrationEmbeddedPublishTest { LOG.info("*** testClientSubscribeBeforeRetainedQoS1IsSent ***"); subscribeToWithQos("/topic", 1); - //super ugly but we need the MQTT client lib finish it's job before us - Thread.sleep(1000); //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, true); @@ -210,9 +205,11 @@ public class ServerIntegrationEmbeddedPublishTest { //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, true); subscribeToWithQos("/topic", 1); + LOG.info("** After subscribe **"); //Verify verifyMessageIsReceivedSuccessfully(); + LOG.info("** Post verify **"); } @Test @@ -220,8 +217,6 @@ public class ServerIntegrationEmbeddedPublishTest { LOG.info("*** testClientSubscribeBeforeNotRetainedQoS2IsSent ***"); subscribeToWithQos("/topic", 2); - //super ugly but we need the MQTT client lib finish it's job before us - Thread.sleep(1000); //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.EXACTLY_ONCE, false); @@ -247,8 +242,6 @@ public class ServerIntegrationEmbeddedPublishTest { LOG.info("*** testClientSubscribeBeforeRetainedQoS2IsSent ***"); subscribeToWithQos("/topic", 2); - //super ugly but we need the MQTT client lib finish it's job before us - Thread.sleep(1000); //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.EXACTLY_ONCE, true); diff --git a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java index d22963cc..937fd310 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java @@ -43,8 +43,8 @@ public class MemorySessionStore implements ISessionsStore { // private Map> m_inflightIDs = new HashMap<>(); //maps clientID->[guid*] private Map> m_enqueuedStore = new HashMap<>(); - //maps clientID->[messageID*] - private Map> m_secondPhaseStore = new HashMap<>(); + //maps clientID->[MessageId -> guid] + private Map> m_secondPhaseStore = new HashMap<>(); private Map> m_messageToGuids; private final IMessagesStore m_messagesStore; @@ -214,17 +214,27 @@ public class MemorySessionStore implements ISessionsStore { } @Override - public void secondPhaseAcknowledged(String clientID, int messageID) { - Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.remove(messageID); + public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID) { + LOG.info("acknowledging inflight clientID <{}> messageID {}", clientID, messageID); + Map m = this.m_inflightStore.get(clientID); + if (m == null) { + LOG.error("Can't find the inFlight record for client <{}>", clientID); + return; + } + String guid = m.remove(messageID); + + LOG.info("Moving to second phase store"); + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); + messageIDs.put(messageID, guid); m_secondPhaseStore.put(clientID, messageIDs); } @Override - public void secondPhaseAckWaiting(String clientID, int messageID) { - Set messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashSet()); - messageIDs.add(messageID); + public String secondPhaseAcknowledged(String clientID, int messageID) { + Map messageIDs = Utils.defaultGet(m_secondPhaseStore, clientID, new HashMap()); + String guid = messageIDs.remove(messageID); m_secondPhaseStore.put(clientID, messageIDs); + return guid; } @Override