Description
Test procedure:
Create a durable subscription to topic "durableunsubtest" with clientid "durableUnsub".
Publish a message with client2 to topic "durableunsubtest". Make sure "retain" flag is not set.
Verify that the message is received (confirming creation of durable subscription).
Unsubscribe from topic "durableunsubtest" from clientid "durableUnsub". Disconnect client.
Publish another message with client2. Make sure "retain" flag is not set.
Connect "durableUnsub" and subscribe to the topic again.
Verify that no messages are received.
JUnit Test Case code:
/*
* Test unsubscribe on a durable subscription topic
*/
@Test(timeout = 60 * 1000)
public void testDurableUnsubscribe() throws Exception {
// Create connection1
final String clientId = "durableUnsub";
MQTT mqtt1 = createMQTTConnection(clientId, false);
mqtt1.setKeepAlive((short) 60);
final BlockingConnection connection1 = mqtt1.blockingConnection();
connection1.connect();
// create a durable subscription for client "durableUnsub"
final String TOPICNAME = "durableunsubtest";
final String payload = "durable unsub test";
Topic[] topic = { new Topic(TOPICNAME, QoS.AT_LEAST_ONCE) };
connection1.subscribe(topic);
// Create connection2 and publish a message on the topic
MQTT mqtt2 = createMQTTConnection(null, true);
mqtt2.setKeepAlive((short) 60);
final BlockingConnection connection2 = mqtt2.blockingConnection();
connection2.connect();
connection2.publish(TOPICNAME, payload.getBytes(), QoS.EXACTLY_ONCE, false);
// Verify that the durable subscription was created by receiving the message
// on connection1
Message message = connection1.receive(10, TimeUnit.SECONDS);
assertNotNull(message);
message.ack();
assertEquals("Unexpected String received", payload, new String(message.getPayload()));
// Unsubscribe the topic on connection1
connection1.unsubscribe(new String[]{TOPICNAME});
// Disconnect connection1
connection1.disconnect();
// Publish another message on connection2 for the same topic
MQTT mqtt2a = createMQTTConnection(null, true);
mqtt2a.setKeepAlive((short) 60);
final BlockingConnection connection2a = mqtt2a.blockingConnection();
connection2a.connect();
connection2a.publish(TOPICNAME, payload.getBytes(), QoS.EXACTLY_ONCE, false);
// Create connection3 with the same clientid "durableUnsub"
// and subscribe to the topic
MQTT mqtt3 = createMQTTConnection(clientId, false);
mqtt3.setKeepAlive((short) 60);
BlockingConnection connection3 = mqtt3.blockingConnection();
connection3.connect();
connection3.subscribe(topic);
message = connection3.receive(10, TimeUnit.SECONDS);
// Since we have unsubscribed before the publish, we should not receive
// any message (retain flag was also not set during publish)
assertNull(message);
}
Attachments
Issue Links
- is duplicated by
-
AMQ-5418 MQTT Subscriber is able to receive messages published after unsubscribe()
-
- Closed
-