[
https://issues.apache.org/jira/browse/AMQ-5358?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
AR updated AMQ-5358:
--------------------
Description:
Durable subscriptions do not work in 5.10 and 5.11-SNAPSHOT.
Test case:
Run default broker.
. Connect "client1" with clean_session=true.
. Subscribe to topic "paho/test"
. Disconnect client1
. Connect "client1" with clean_session=false
. Subscribe to topic "paho/test"
. Disconnect "client1"
. Connect "client2"
. Publish message "hello world" to topic "paho/test"
. Disconnect "client2"
. Connect "client1" with clean_session=false
. Subscribe to topic "paho/test"
. Verify that the message is received.
Junit test code attached.
was:
Durable subscriptions do not work in 5.10 and 5.11-SNAPSHOT.
Test case:
Run default broker.
. Connect "client1" with clean_session=true.
. Subscribe to topic "paho/test"
. Disconnect client1
. Connect "client1" with clean_session=false
. Subscribe to topic "paho/test"
. Disconnect "client1"
. Connect "client2"
. Publish message "hello world" to topic "paho/test"
. Disconnect "client2"
. Connect "client1" with clean_session=false
. Subscribe to topic "paho/test"
. Verify that the message is received.
Here is the Junit test code:
----------------------------------
package com.mytests.activemqtests;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
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.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.junit.Test;
import junit.framework.TestCase;
public class MqttDurableSubTest extends TestCase implements MqttCallback {
final static String BROKER1_URL = "tcp://localhost:1883";
final static int MQTT_QOS_ATMOSTONCE = 0;
final static int MQTT_QOS_ATLEASTONCE = 1;
final static int MQTT_QOS_EXACTLYONCE = 2;
private boolean mMessageReceived = false;
private CountDownLatch mLatch;
public MqttDurableSubTest(String name) {
super(name);
}
protected void setUp() throws Exception {
super.setUp();
}
protected void tearDown() throws Exception {
super.tearDown();
}
@Test
public void testDuplicateMsg() throws Exception {
MqttConnectOptions client1ConnOpt = new MqttConnectOptions();
MqttClient client1 = new MqttClient(BROKER1_URL, "client1");
client1ConnOpt.setCleanSession(true);
client1.connect();
System.out.println("Client1 Connected to " + BROKER1_URL);
client1.subscribe("paho/test", MQTT_QOS_ATMOSTONCE);
Thread.sleep(2000);
client1.disconnect();
System.out.println("Client1 completed cleansession=1
subscription.");
client1ConnOpt.setCleanSession(false);
client1.setCallback(this);
client1.connect(client1ConnOpt);
System.out.println("Client1 Connected to " + BROKER1_URL);
client1.subscribe("paho/test", MQTT_QOS_ATMOSTONCE);
Thread.sleep(2000);
client1.disconnect();
System.out.println("Client1 completed durable subscription.");
MqttConnectOptions client2ConnOpt = new MqttConnectOptions();
client1ConnOpt.setCleanSession(true);
MqttClient client2 = new MqttClient(BROKER1_URL, "client2");
client2.setCallback(this);
client2.connect(client2ConnOpt);
client2.publish("paho/test", "hello world".getBytes(),
MQTT_QOS_ATMOSTONCE, false);
System.out.println("Client2 published");
client2.disconnect();
client1ConnOpt.setCleanSession(false);
client1.setCallback(this);
client1.connect(client1ConnOpt);
System.out.println("Client1 Connected to " + BROKER1_URL);
client1.subscribe("paho/test", MQTT_QOS_ATMOSTONCE);
mLatch = new CountDownLatch(1);
mLatch.await(10, TimeUnit.SECONDS);
assertEquals("Success!", true, mMessageReceived);
System.out.println("Done");
}
@Override
public void connectionLost(Throwable arg0) {
// TODO Auto-generated method stub
}
@Override
public void deliveryComplete(IMqttDeliveryToken arg0) {
// TODO Auto-generated method stub
}
@Override
public void messageArrived(String topic, MqttMessage message) throws
Exception {
System.out.println("MQTT - messageArrived "+topic+" , Message:
["+message+"] , ReceviedTime: ["+ new Date() +"] , QoS: ["+message.getQos()+"]
, isDup: ["+message.isDuplicate()+"]" );
if (message.toString().equals("hello world")) {
mMessageReceived = true;
mLatch.countDown();
}
}
}
> MQTT Durable subscription broken in 5.10 and 5.11
> -------------------------------------------------
>
> Key: AMQ-5358
> URL: https://issues.apache.org/jira/browse/AMQ-5358
> Project: ActiveMQ
> Issue Type: Bug
> Components: MQTT
> Affects Versions: 5.10.0, 5.11.0
> Environment: Mac OS X, MQTT Paho client library version 0.4.0
> Reporter: AR
> Attachments: MqttDurableSubTest.java
>
>
> Durable subscriptions do not work in 5.10 and 5.11-SNAPSHOT.
> Test case:
> Run default broker.
> . Connect "client1" with clean_session=true.
> . Subscribe to topic "paho/test"
> . Disconnect client1
> . Connect "client1" with clean_session=false
> . Subscribe to topic "paho/test"
> . Disconnect "client1"
> . Connect "client2"
> . Publish message "hello world" to topic "paho/test"
> . Disconnect "client2"
> . Connect "client1" with clean_session=false
> . Subscribe to topic "paho/test"
> . Verify that the message is received.
> Junit test code attached.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
