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

* Added integration test to check QoS downgrading for messages delivered to registered subscribers

Browse Source 
* Fixed a deouble delivery of messages in case of QoS2 publish
This commit is contained in:
andrea
2013-12-26 18:32:41 +01:00
parent 2c28c0b771
commit 33705c52e0
4 changed files with 265 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/pom.xml
+4 -4
View File
@@ -36,6 +36,7 @@
<groupId>com.googlecode.disruptor</groupId>
<artifactId>disruptor</artifactId>
<version>2.7.1</version>
<!--<version>3.2.0</version>-->
</dependency>
<dependency>
@@ -101,10 +102,9 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<excludes>
<exclude>org/dna/mqtt/moquette/proto/TestUtils.java</exclude>
<exclude>org/dna/mqtt/moquette/proto/TestUtils$*</exclude>
</excludes>
<includes>
<include>**/*Test.java</include>
</includes>
</configuration>
</plugin>
broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+18 -6
View File
@@ -200,7 +200,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
}
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<ValueEvent> {
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<ValueEvent> {
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<ValueEvent> {
}
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<ValueEvent> {
}
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<ValueEvent> {
* 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<ValueEvent> {
}
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<ValueEvent> {
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);
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationSuite.java
+231
View File
@@ -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());
}
}