From 33705c52e02305e1b03e7ee4687b76a3370bd7b8 Mon Sep 17 00:00:00 2001 From: andrea Date: Thu, 26 Dec 2013 18:32:41 +0100 Subject: [PATCH] * Added integration test to check QoS downgrading for messages delivered to registered subscribers * Fixed a deouble delivery of messages in case of QoS2 publish --- ChangeLog.txt | 21 +- broker/pom.xml | 8 +- .../messaging/spi/impl/ProtocolProcessor.java | 24 +- .../ServerIntegrationQoSValidationSuite.java | 231 ++++++++++++++++++ 4 files changed, 265 insertions(+), 19 deletions(-) create mode 100644 broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationSuite.java diff --git a/ChangeLog.txt b/ChangeLog.txt index 3c2ced9b..48a24779 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,13 +1,16 @@ -Version 0.2: - Fixed issues (10, 11, 12, 13) - Fixed interdependency bug in test - -Version 0.3: - Added Netty IO connectors - Dropped tests based on Fuse source client lib moving to Paho java client - Version 0.4: Dropped support for MINA Fix Issue 15 (bad re-activation of client's subscriptions after a recconection) Fix Issue 16 (bad re-publish of stored messages after a client reconnection) - Fixed a bug in unsubscribe due to missed clientID check \ No newline at end of file + Fixed a bug in unsubscribe due to missed clientID check + Fixed Issue 18 (descriptor of a disconnected (not active) client) + Bug fix for override subscriptions to the same /topic but with different QoS (latest wins) + Added test to check the downgrading of QoS of messages sent to registered subscribers (fixed a bug of double delivery of QoS2 messages) + +Version 0.3: + Added Netty IO connectors + Dropped tests based on Fuse source client lib moving to Paho java client + +Version 0.2: + Fixed issues (10, 11, 12, 13) + Fixed interdependency bug in test \ No newline at end of file diff --git a/broker/pom.xml b/broker/pom.xml index 14635ee8..07ca44c4 100644 --- a/broker/pom.xml +++ b/broker/pom.xml @@ -36,6 +36,7 @@ com.googlecode.disruptor disruptor 2.7.1 + @@ -101,10 +102,9 @@ org.apache.maven.plugins maven-surefire-plugin - - org/dna/mqtt/moquette/proto/TestUtils.java - org/dna/mqtt/moquette/proto/TestUtils$* - + + **/*Test.java + 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 1aef5991..e5cc1ecd 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 @@ -200,7 +200,7 @@ class ProtocolProcessor implements EventHandler { } protected void processPublish(PublishEvent evt) { - LOG.trace("processPublish invoked with {}", evt); + LOG.trace("PUB --PUBLISH--> SRV processPublish invoked with {}", evt); final String topic = evt.getTopic(); final AbstractMessage.QOSType qos = evt.getQos(); final ByteBuffer message = evt.getMessage(); @@ -221,7 +221,10 @@ class ProtocolProcessor implements EventHandler { sendPubRec(evt.getClientID(), evt.getMessageID()); } - publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + //NB publish 2 subscribers for QoS 2 happen upon PUBREL from publsher + if (qos != AbstractMessage.QOSType.EXACTLY_ONCE) { + publish2Subscribers(topic, qos, message, retain, evt.getMessageID()); + } if (qos == AbstractMessage.QOSType.LEAST_ONE) { if (publishKey == null) { @@ -247,6 +250,10 @@ class ProtocolProcessor implements EventHandler { LOG.debug("subscription tree {}", subscriptions.dumpTree()); } for (final Subscription sub : subscriptions.matches(topic)) { + if (qos.ordinal() > sub.getRequestedQos().ordinal()) { + qos = sub.getRequestedQos(); + } + LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}", sub.getClientId(), sub.getTopic(), qos, sub.isActive()); @@ -275,7 +282,9 @@ class ProtocolProcessor implements EventHandler { } private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) { - sendPublish(clientId, topic, qos, message, retained, 0); + //TODO pay attention to the message ID can't be 0 and it's the message sent to subscriber + int messageID = 1; + sendPublish(clientId, topic, qos, message, retained, messageID); } private void sendPublish(String clientId, String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained, int messageID) { @@ -309,7 +318,7 @@ class ProtocolProcessor implements EventHandler { } private void sendPubRec(String clientID, int messageID) { - LOG.trace("sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); + LOG.trace("PUB <--PUBREC-- SRV sendPubRec invoked for clientID {} with messageID {}", clientID, messageID); PubRecMessage pubRecMessage = new PubRecMessage(); pubRecMessage.setMessageID(messageID); @@ -346,6 +355,7 @@ class ProtocolProcessor implements EventHandler { * to all interested subscribers. * */ void processPubRel(String clientID, int messageID) { + LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID); String publishKey = String.format("%s%d", clientID, messageID); PublishEvent evt = m_storageService.retrieveQoS2Message(publishKey); @@ -364,7 +374,7 @@ class ProtocolProcessor implements EventHandler { } private void sendPubComp(String clientID, int messageID) { - LOG.debug("sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID); + LOG.debug("PUB <--PUBCOMP-- SRV sendPubComp invoked for clientID {} ad messageID {}", clientID, messageID); PubCompMessage pubCompMessage = new PubCompMessage(); pubCompMessage.setMessageID(messageID); @@ -374,15 +384,17 @@ class ProtocolProcessor implements EventHandler { void processPubRec(String clientID, int messageID) { //once received a PUBREC reply with a PUBREL(messageID) - LOG.debug("processPubRec invoked for clientID {} ad messageID {}", clientID, messageID); + LOG.debug("\t\tSRV <--PUBREC-- SUB processPubRec invoked for clientID {} ad messageID {}", clientID, messageID); PubRelMessage pubRelMessage = new PubRelMessage(); pubRelMessage.setMessageID(messageID); + pubRelMessage.setQos(AbstractMessage.QOSType.LEAST_ONE); // m_clientIDs.get(clientID).getSession().write(pubRelMessage); disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage)); } void processPubComp(String clientID, int messageID) { + 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 String publishKey = String.format("%s%d", clientID, messageID); m_storageService.cleanInFlight(publishKey); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationSuite.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationSuite.java new file mode 100644 index 00000000..7c12667a --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationSuite.java @@ -0,0 +1,231 @@ +package org.dna.mqtt.moquette.server; + + +import java.io.File; +import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.eclipse.paho.client.mqttv3.IMqttClient; +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; +import org.eclipse.paho.client.mqttv3.MqttCallback; +import org.eclipse.paho.client.mqttv3.MqttClient; +import org.eclipse.paho.client.mqttv3.MqttClientPersistence; +import org.eclipse.paho.client.mqttv3.MqttMessage; +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; +import org.junit.After; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import org.junit.Before; +import org.junit.Test; +import org.junit.BeforeClass; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class ServerIntegrationQoSValidationSuite { + static class TestCallback implements MqttCallback { + + private MqttMessage m_message; + private String m_topic; + private CountDownLatch m_latch = new CountDownLatch(1); + + public MqttMessage getMessage(boolean checkElapsed) { + try { + boolean elapsed = !m_latch.await(1, TimeUnit.SECONDS); + if (elapsed && checkElapsed) { + throw new IllegalStateException("Elapsed the timeout to get the result"); + } + } catch (InterruptedException e) { + e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + } + return m_message; + } + + public String getTopic() { + try { + m_latch.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + } + return m_topic; + } + + void reinit() { + m_latch = new CountDownLatch(1); + m_message = null; + m_topic = null; + } + + public void connectionLost(Throwable cause) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void messageArrived(String topic, MqttMessage message) throws Exception { + m_message = message; + m_topic = topic; + m_latch.countDown(); + } + + public void deliveryComplete(IMqttDeliveryToken token) { + //To change body of implemented methods use File | Settings | File Templates. + } + } + + + private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); + + Server m_server; + static MqttClientPersistence s_subDataStore; + static MqttClientPersistence s_pubDataStore; + + IMqttClient m_subscriber; + IMqttClient m_publisher; + TestCallback m_callback; + + @BeforeClass + public static void beforeTests() { + String tmpDir = System.getProperty("java.io.tmpdir"); + s_subDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriber"); + s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); + } + + protected void startServer() throws IOException { + m_server = new Server(); + m_server.startServer(); + } + + @Before + public void setUp() throws Exception { + startServer(); + + m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_subDataStore); + m_callback = new TestCallback(); + m_subscriber.setCallback(m_callback); + m_subscriber.connect(); + + m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); +// m_callback = new TestCallback(); +// m_subscriber.setCallback(m_callback); + m_publisher.connect(); + } + + @After + public void tearDown() throws Exception { + if (m_publisher.isConnected()) { + m_publisher.disconnect(); + } + + if (m_subscriber.isConnected()) { + m_subscriber.disconnect(); + } + + m_server.stopServer(); + File dbFile = new File(Server.STORAGE_FILE_PATH); + if (dbFile.exists()) { + dbFile.delete(); + } + } + + @Test + public void checkSubscriberQoS0ReceiveQoS0publishes() throws Exception { + LOG.info("*** checkSubscriberQoS0ReceiveQoS0publishes ***"); + m_subscriber.subscribe("/topic", 0); + + m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS0", message.toString()); + assertEquals(0, message.getQos()); + } + + @Test + public void checkSubscriberQoS0ReceiveQoS1publishes_downgrade() throws Exception { + LOG.info("*** checkSubscriberQoS0ReceiveQoS1publishes_downgrade ***"); + m_subscriber.subscribe("/topic", 0); + + m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS1", message.toString()); + assertEquals(0, message.getQos()); + } + + @Test + public void checkSubscriberQoS0ReceiveQoS2publishes_downgrade() throws Exception { + LOG.info("*** checkSubscriberQoS0ReceiveQoS2publishes_downgrade ***"); + m_subscriber.subscribe("/topic", 0); + + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS2", message.toString()); + assertEquals(0, message.getQos()); + } + + @Test + public void checkSubscriberQoS1ReceiveQoS0publishes() throws Exception { + LOG.info("*** checkSubscriberQoS1ReceiveQoS0publishes ***"); + m_subscriber.subscribe("/topic", 1); + + m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS0", message.toString()); + assertEquals(0, message.getQos()); + } + + @Test + public void checkSubscriberQoS1ReceiveQoS1publishes() throws Exception { + LOG.info("*** checkSubscriberQoS1ReceiveQoS1publishes ***"); + m_subscriber.subscribe("/topic", 1); + + m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS1", message.toString()); + assertEquals(1, message.getQos()); + } + + @Test + public void checkSubscriberQoS1ReceiveQoS2publishes_downgrade() throws Exception { + LOG.info("*** checkSubscriberQoS1ReceiveQoS2publishes_downgrade ***"); + m_subscriber.subscribe("/topic", 1); + + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS2", message.toString()); + assertEquals(1, message.getQos()); + } + + @Test + public void checkSubscriberQoS2ReceiveQoS0publishes() throws Exception { + LOG.info("*** checkSubscriberQoS2ReceiveQoS0publishes ***"); + m_subscriber.subscribe("/topic", 2); + + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 0, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS2", message.toString()); + assertEquals(0, message.getQos()); + } + + @Test + public void checkSubscriberQoS2ReceiveQoS1publishes() throws Exception { + LOG.info("*** checkSubscriberQoS2ReceiveQoS1publishes ***"); + m_subscriber.subscribe("/topic", 2); + + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 1, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS2", message.toString()); + assertEquals(1, message.getQos()); + } + + @Test + public void checkSubscriberQoS2ReceiveQoS2publishes() throws Exception { + LOG.info("*** checkSubscriberQoS2ReceiveQoS2publishes ***"); + m_subscriber.subscribe("/topic", 2); + + m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); + MqttMessage message = m_callback.getMessage(true); + assertEquals("Hello world MQTT QoS2", message.toString()); + assertEquals(2, message.getQos()); + } + +}