Apache/activemq
1
0
Fork 0
mirror of https://github.com/apache/activemq.git synced 2025-02-09 11:35:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-5511 - retained message for zero-byte clientId client

Browse Source
This commit is contained in:
Dejan Bosanac 2015-01-09 12:50:39 +01:00
parent f303f85ce3
commit 7948d69056
2 changed files with 28 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/AbstractMQTTSubscriptionStrategy.java
View File

@ -129,7 +129,11 @@ public abstract class AbstractMQTTSubscriptionStrategy implements MQTTSubscripti
// use actual client id used to create connection to lookup connection // use actual client id used to create connection to lookup connection
// context // context
final String connectionInfoClientId = protocol.getClientId(); String connectionInfoClientId = protocol.getClientId();
// for zero-byte client ids we used connection id
if (connectionInfoClientId == null || connectionInfoClientId.isEmpty()) {
connectionInfoClientId = protocol.getConnectionId().toString();
}
final ConnectionContext connectionContext = regionBroker.getConnectionContext(connectionInfoClientId); final ConnectionContext connectionContext = regionBroker.getConnectionContext(connectionInfoClientId);
// get all matching Topics // get all matching Topics

23
activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
View File

@ -966,6 +966,29 @@ public class MQTTTest extends MQTTTestSupport {
newConnection.disconnect(); newConnection.disconnect();
} }
@Test(timeout = 60 * 1000)
public void testNoClientId() throws Exception {
final MQTT mqtt = createMQTTConnection("", true);
final BlockingConnection connection = mqtt.blockingConnection();
connection.connect();
Wait.waitFor(new Wait.Condition() {
@Override
public boolean isSatisified() throws Exception {
return connection.isConnected();
}
});
connection.subscribe(new Topic[]{new Topic("TopicA", QoS.AT_LEAST_ONCE)});
connection.publish("TopicA", "test".getBytes(), QoS.AT_LEAST_ONCE, true);
Message message = connection.receive(3, TimeUnit.SECONDS);
assertNotNull(message);
Thread.sleep(2000);
connection.subscribe(new Topic[]{new Topic("TopicA", QoS.AT_LEAST_ONCE)});
//TODO fix audit problem for retained messages
//message = connection.receive(3, TimeUnit.SECONDS);
//assertNotNull(message);
}
@Test(timeout = 60 * 1000) @Test(timeout = 60 * 1000)
public void testCleanSession() throws Exception { public void testCleanSession() throws Exception {
final String CLIENTID = "cleansession"; final String CLIENTID = "cleansession";