diff --git a/ChangeLog.txt b/ChangeLog.txt index 39fa7543..21031042 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -4,6 +4,8 @@ Version 0.5: Created a self contained distribution package (issue 22) Fix issue 28, OSGi-fied the broker Fixed issue #31 missed to insert the hawtdb jars + Added a separate mvn module to host the OSGi bundle declaration + Fixed issue #30 Version 0.4: Dropped support for MINA @@ -20,4 +22,4 @@ Version 0.3: Version 0.2: Fixed issues (10, 11, 12, 13) - Fixed interdependency bug in test \ No newline at end of file + Fixed interdependency bug in test diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index aaade00e..d3f6d83e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -261,9 +261,6 @@ class ProtocolProcessor implements EventHandler { //QoS 0 sendPublish(sub.getClientId(), topic, qos, message, false); } else { - //TODO BUG if the sub is not active the code and the cleanSession is true - //then publish to the subscriber (that is not active) - //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it if (!sub.isCleanSession() && !sub.isActive()) { @@ -278,7 +275,9 @@ class ProtocolProcessor implements EventHandler { m_storageService.addInFlight(newPublishEvt, publishKey); } //publish - sendPublish(sub.getClientId(), topic, qos, message, false); + if (sub.isActive()) { + sendPublish(sub.getClientId(), topic, qos, message, false); + } } } } diff --git a/broker/src/main/resources/log4j.properties b/broker/src/main/resources/log4j.properties index db932b00..970a1215 100644 --- a/broker/src/main/resources/log4j.properties +++ b/broker/src/main/resources/log4j.properties @@ -12,15 +12,23 @@ log4j.rootLogger=ERROR, stdout, file log4j.logger.org.dna.mqtt=INFO log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationFuseTest=DEBUG log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=INFO +#log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=DEBUG +#log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=TRACE log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging=WARN #Protocol parsing log4j.logger.org.dna.mqtt.moquette.server.netty.NettyMQTTHandler=WARN +#log4j.logger.org.dna.mqtt.moquette.server.netty.NettyMQTTHandler=DEBUG log4j.logger.org.dna.mqtt.moquette.parser.netty=WARN +#Storage server +#log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore=DEBUG +#log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService=DEBUG # stdout appender is set to be a ConsoleAppender. log4j.appender.stdout=org.apache.log4j.ConsoleAppender -#log4j.appender.stdout.Threshold=WARN -log4j.appender.stdout.Threshold=INFO +log4j.appender.stdout.Threshold=WARN +#log4j.appender.stdout.Threshold=INFO +#log4j.appender.stdout.Threshold=DEBUG +#log4j.appender.stdout.Threshold=TRACE log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 09ce77ec..f3b01543 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -22,6 +22,7 @@ import org.dna.mqtt.moquette.server.ServerChannel; import static org.junit.Assert.*; import static org.mockito.Mockito.*; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; /** @@ -273,7 +274,7 @@ public class ProtocolProcessorTest { } @Test - public void processPublishNoPublishToInactiveSubscriptions() { + public void publishNoPublishToInactiveSubscriptions() { SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE, false); inactiveSub.setActive(false); @@ -291,4 +292,23 @@ public class ProtocolProcessorTest { assertNull(m_receivedMessage); } + + @Test + public void publishToAnInactiveSubscriptionsCleanSession() { + SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); + Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE, true); + inactiveSub.setActive(false); + List inactiveSubscriptions = Arrays.asList(inactiveSub); + when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); + m_processor = new ProtocolProcessor(); + m_processor.init(mockedSubscriptions, m_storageService); + + //Exercise + ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); + PublishEvent pubEvt = new PublishEvent("/topic", AbstractMessage.QOSType.MOST_ONE, buffer, true, "Publisher", null); + m_processor.processPublish(pubEvt); + + //Verify no message is received + assertNull(m_receivedMessage); + } }