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 d8af6c4f..841abc66 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java +++ b/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. diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index 6f9a9a0b..de4d076c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/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 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); } 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 499a0c0d..67077d16 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 @@ -142,7 +142,7 @@ class ProtocolProcessor implements EventHandler { //init the output ringbuffer m_executor = Executors.newFixedThreadPool(1); - /*Disruptor*/ m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); + /*Disruptor*/ 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 { 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!! - m_messagesStore.addInFlight(publishEvt, clientID, messageID); + //TODO implement inFlight!! + 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 { //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 { 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) diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java index 17777fb8..45295a3d 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java @@ -101,7 +101,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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 m_disruptor = new Disruptor(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor, ProducerType.MULTI, new BusySpinWaitStrategy());*/ m_disruptor.handleEventsWith(this); 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 9e2f7d37..2455db4f 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 @@ -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 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); diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/MapDBPersistentStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/MapDBPersistentStoreTest.java index 4c4497b4..1b50cfbb 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/MapDBPersistentStoreTest.java +++ b/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"); diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index 38274c8a..920607e8 100644 --- a/broker/src/test/resources/log4j.properties +++ b/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