From f649ebff24881a44defab1b6f9056a4ea8b2e1c8 Mon Sep 17 00:00:00 2001 From: andrea Date: Mon, 25 Nov 2013 18:53:50 +0100 Subject: [PATCH] Fixed issue #15, missed notification of publish due to bad re-activation on a nested topic tree. --- ChangeLog.txt | 1 + .../spi/impl/subscriptions/TreeNode.java | 2 +- broker/src/main/resources/log4j.properties | 7 ++- .../server/ServerIntegrationPahoTest.java | 27 ++++++++++- tools_scripts/issue15_receiver.groovy | 48 +++++++++++++++++++ tools_scripts/issue15_sender.groovy | 27 +++++++++++ 6 files changed, 108 insertions(+), 4 deletions(-) create mode 100644 tools_scripts/issue15_receiver.groovy create mode 100644 tools_scripts/issue15_sender.groovy diff --git a/ChangeLog.txt b/ChangeLog.txt index b756c485..d9822c3f 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -8,3 +8,4 @@ Version 0.3: Version 0.4: Dropped support for MINA + Fix Issue 15 (bad re-activation of client's subscriptions after a recconection) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java index e905d9ee..1dcbc2ef 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java @@ -148,7 +148,7 @@ class TreeNode { //go deep for (TreeNode child : m_children) { - child.deactivate(clientID); + child.activate(clientID); } } diff --git a/broker/src/main/resources/log4j.properties b/broker/src/main/resources/log4j.properties index 1b51ddd9..db932b00 100644 --- a/broker/src/main/resources/log4j.properties +++ b/broker/src/main/resources/log4j.properties @@ -12,10 +12,15 @@ 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.SimpleMessaging=WARN +#Protocol parsing +log4j.logger.org.dna.mqtt.moquette.server.netty.NettyMQTTHandler=WARN +log4j.logger.org.dna.mqtt.moquette.parser.netty=WARN # stdout appender is set to be a ConsoleAppender. log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.Threshold=WARN +#log4j.appender.stdout.Threshold=WARN +log4j.appender.stdout.Threshold=INFO 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/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java index 7703813b..151fb330 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java @@ -62,6 +62,8 @@ public class ServerIntegrationPahoTest { private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); static MqttClientPersistence s_dataStore; + static MqttClientPersistence s_pubDataStore; + Server m_server; IMqttClient m_client; @@ -71,6 +73,7 @@ public class ServerIntegrationPahoTest { public static void beforeTests() { String tmpDir = System.getProperty("java.io.tmpdir"); s_dataStore = new MqttDefaultFilePersistence(tmpDir); + s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); } protected void startServer() throws IOException { @@ -275,9 +278,29 @@ public class ServerIntegrationPahoTest { assertEquals("Hello MQTT", m_callback.getMessage().toString()); } - + + @Test + public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws Exception { + LOG.info("*** checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession ***"); + MqttConnectOptions options = new MqttConnectOptions(); + options.setCleanSession(false); + m_client.connect(options); + m_client.subscribe("/topic", 1); + m_client.disconnect(); + + m_client.connect(options); + m_client.subscribe("/topic", 1); + + //publish a QoS 1 message another client publish a message on the topic + publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1); + + //Verify that after a reconnection the client receive the message + assertNotNull(m_callback.getMessage()); + assertEquals("Hello MQTT", m_callback.getMessage().toString()); + } + private void publishFromAnotherClient(String topic, byte[] payload, int qos) throws Exception { - IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_dataStore); + IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_pubDataStore); anotherClient.connect(); anotherClient.publish(topic, payload, qos, false); anotherClient.disconnect(); diff --git a/tools_scripts/issue15_receiver.groovy b/tools_scripts/issue15_receiver.groovy new file mode 100644 index 00000000..a2900e2b --- /dev/null +++ b/tools_scripts/issue15_receiver.groovy @@ -0,0 +1,48 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import java.net.URISyntaxException + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +class Reciever { + MQTT mqtt = new MQTT() + BlockingConnection connection + + public void connectAndSubscribe(boolean cleanSession) { + mqtt.setCleanSession(cleanSession) + mqtt.setHost("localhost", 1883) + + mqtt.setClientId("reciever") + connection = mqtt.blockingConnection() + connection.connect() + Topic[] topics = [new Topic("/topic", QoS.AT_LEAST_ONCE)] + byte[] qoses = connection.subscribe(topics) + } +} + + +Reciever r = new Reciever() +r.connectAndSubscribe(false) +Message message +3.times { i -> + message = r.connection.receive() + println "Topic: ${message.topic}, payload: ${new String(message.payload)}" + message.ack() +} +r.connection.kill() +//r.connection.disconnect() + +Thread.sleep(3000) + +r.connectAndSubscribe(false) +3.times { i -> + message = r.connection.receive() + println "Topic: ${message.topic}, payload: ${new String(message.payload)}" + message.ack() +} +r.connection.disconnect() + diff --git a/tools_scripts/issue15_sender.groovy b/tools_scripts/issue15_sender.groovy new file mode 100644 index 00000000..f79d4a36 --- /dev/null +++ b/tools_scripts/issue15_sender.groovy @@ -0,0 +1,27 @@ +//@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/') +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import java.net.URISyntaxException + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.QoS + + +MQTT mqtt = new MQTT() +//mqtt.setHost("test.mosquitto.org", 1883); +mqtt.setHost("localhost", 1883) + +BlockingConnection connection = mqtt.blockingConnection() +connection.connect() + +//for (int i = 0; i < 1000; i++) { +(1..1000).each{ i -> +// LOG.info("Publishing"); + String payload = "Hello world MQTT!!${i}" + connection.publish("/topic", payload.getBytes(), QoS.AT_LEAST_ONCE, false) + println payload + Thread.sleep 1000 +} + +connection.disconnect()