anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed issue '30 to avoid publish of QoS0 to not Active clean session subscriptions

Browse Source
This commit is contained in:
selva.andre
2014-02-08 20:01:59 +01:00
parent 56a178c786
commit 4ff6b2aeae
4 changed files with 37 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+3 -4
View File
@@ -261,9 +261,6 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
//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<ValueEvent> {
m_storageService.addInFlight(newPublishEvt, publishKey);
}
//publish
sendPublish(sub.getClientId(), topic, qos, message, false);
if (sub.isActive()) {
sendPublish(sub.getClientId(), topic, qos, message, false);
}
}
}
}
broker/src/main/resources/log4j.properties
+10 -2
View File
@@ -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
broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java
+21 -1
View File
@@ -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<Subscription> 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);
}
}