Renamed addInFlight and cleanInFlight to be related to temp messages, not really relate to in-flight

broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java

@@ -81,9 +81,9 @@ public interface IMessagesStore {
 
     void dropMessagesInSession(String clientID);
 
-    void cleanInFlight(String clientID, int packetID);
+    void cleanTemporaryPublish(String clientID, int packetID);
 
-    void addInFlight(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/impl/MemoryStorageService.java

@@ -127,7 +127,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
     }
 
     @Override
-    public void cleanInFlight(String clientID, int packetID) {
+    public void cleanTemporaryPublish(String clientID, int packetID) {
         String publishKey = String.format("%s%d", clientID, packetID);
         m_inflightStore.remove(publishKey);
         Set<Integer> inFlightForClient = m_inflightIDs.get(clientID);
@@ -137,7 +137,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {
     }
 
     @Override
-    public void addInFlight(PublishEvent evt, String clientID, int packetID) {
+    public void storeTemporaryPublish(PublishEvent evt, String clientID, int packetID) {
         String publishKey = String.format("%s%d", clientID, packetID);
         m_inflightStore.put(publishKey, evt);
     }

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -142,7 +142,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         //init the output ringbuffer
         m_executor = Executors.newFixedThreadPool(1);
 
-        /*Disruptor<ValueEvent>*/ m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
+        /*Disruptor<ValueEvent>*/ m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); //128 to break the broker
         m_disruptor.handleEventsWith(this);
         m_disruptor.start();
 
@@ -332,10 +332,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         if (qos == AbstractMessage.QOSType.MOST_ONE) { //QoS0
             forward2Subscribers(publishEvt);
         } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1
-            //TODO use a message store for TO PUBLISH MESSAGES it has nothing to do with inFlight!!
+            //TODO implement inFlight!!
-            m_messagesStore.addInFlight(publishEvt, clientID, messageID);
+            m_messagesStore.storeTemporaryPublish(publishEvt, clientID, messageID);
             forward2Subscribers(publishEvt);
-            m_messagesStore.cleanInFlight(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);
@@ -414,7 +414,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
                     //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.addInFlight(newPublishEvt, sub.getClientId(), messageID);
+                        m_messagesStore.storeTemporaryPublish(newPublishEvt, sub.getClientId(), messageID);
                     }
                     //publish
                     if (sub.isActive()) {
@@ -552,7 +552,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         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.cleanInFlight(clientID, messageID);
+        m_messagesStore.cleanTemporaryPublish(clientID, messageID);
     }
     
     @MQTTMessage(message = DisconnectMessage.class)

broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java

@@ -101,7 +101,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     public void init(IConfig configProps) {
         subscriptions = new SubscriptionsStore();
         m_executor = Executors.newFixedThreadPool(1);
-        m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 128, m_executor); //1024 * 32
+        m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); //128 to break the broker
         /*Disruptor<ValueEvent> m_disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor,
                 ProducerType.MULTI, new BusySpinWaitStrategy());*/
         m_disruptor.handleEventsWith(this);

broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java

@@ -188,7 +188,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
 
     //----------------- In flight methods -----------------
     @Override
-    public void cleanInFlight(String clientID, int packetID) {
+    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);
@@ -198,7 +198,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
     }
 
     @Override
-    public void addInFlight(PublishEvent evt, String clientID, int packetID) {
+    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);

broker/src/test/java/org/eclipse/moquette/spi/impl/MapDBPersistentStoreTest.java

@@ -96,7 +96,7 @@ public class MapDBPersistentStoreTest {
         assertEquals(1, packetId);
 
         //release the ID
-        m_storageService.cleanInFlight("CLIENT", packetId);
+        m_storageService.cleanTemporaryPublish("CLIENT", packetId);
 
         //request a second packetID, counter restarts from 0
         packetId = m_storageService.nextPacketID("CLIENT");

broker/src/test/resources/log4j.properties

@@ -1,24 +1,19 @@
 # Set root logger level to DEBUG and its only appender to A1.
 log4j.rootLogger=ERROR, stdout, file
 
-#log4j.logger.org.eclipse.moquette=WARN
+#og4j.logger.org.eclipse.moquette=WARN
 log4j.logger.org.eclipse.moquette=DEBUG
-#log4j.logger.org.eclipse.moquette.server.ServerIntegrationFuseTest=DEBUG
-#log4j.logger.org.eclipse.moquette.server.ServerLowlevelMessagesIntegrationTests=DEBUG
-log4j.logger.org.eclipse.moquette.server.ServerIntegrationQoSValidationTest=INFO
 
-#log4j.logger.org.eclipse.moquette.spi.impl.ProtocolProcessor=WARN
 log4j.logger.org.eclipse.moquette.spi.impl.ProtocolProcessor=DEBUG
-log4j.logger.org.eclipse.moquette.spi.impl.SimpleMessaging=WARN
+log4j.logger.org.eclipse.moquette.spi.impl.SimpleMessaging=DEBUG
 
 #Protocol parsing
-log4j.logger.org.eclipse.moquette.server.netty.NettyMQTTHandler=WARN
+log4j.logger.org.eclipse.moquette.server.netty.NettyMQTTHandler=DEBUG
-#log4j.logger.org.eclipse.moquette.server.netty.NettyMQTTHandler=DEBUG
 log4j.logger.org.eclipse.moquette.parser.netty=DEBUG
 
 #Storage service
 #log4j.logger.org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore=DEBUG
-log4j.logger.org.eclipse.moquette.spi.persistence.MapDBPersistentStore=WARN
+log4j.logger.org.eclipse.moquette.spi.persistence.MapDBPersistentStore=DEBUG
 
 # stdout appender is set to be a ConsoleAppender.
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender