Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-5734 

Support MQTT 3.1 silent subscription fail
This commit is contained in:
Dejan Bosanac 2015-04-20 18:15:20 +02:00
parent 4a821186a4
commit a4fbe70872
3 changed files with 36 additions and 2 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/MQTTProtocolConverter.java
View File

@ -87,6 +87,8 @@ public class MQTTProtocolConverter {
private static final Logger LOG = LoggerFactory.getLogger(MQTTProtocolConverter.class);
public static final String QOS_PROPERTY_NAME = "ActiveMQ.MQTT.QoS";
public static final int V3_1 = 3;
public static final int V3_1_1 = 4;
private static final IdGenerator CONNECTION_ID_GENERATOR = new IdGenerator();
private static final MQTTFrame PING_RESP_FRAME = new PINGRESP().encode();
@ -119,6 +121,8 @@ public class MQTTProtocolConverter {
private final MQTTPacketIdGenerator packetIdGenerator;
private boolean publishDollarTopics;
public int version;
private final FactoryFinder STRATAGY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/strategies/");
/*
@ -246,6 +250,8 @@ public class MQTTProtocolConverter {
passswd = connect.password().toString();
}
version = connect.version();
configureInactivityMonitor(connect.keepAlive());
connectionInfo.setConnectionId(connectionId);

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

@ -204,7 +204,13 @@ public abstract class AbstractMQTTSubscriptionStrategy implements MQTTSubscripti
if (response.isException()) {
final Throwable throwable = ((ExceptionResponse) response).getException();
LOG.warn("Error subscribing to {}", topicName, throwable);
qos[0] = SUBSCRIBE_ERROR;
// version 3.1 don't supports silent fail
// version 3.1.1 send "error" qos
if (protocol.version == protocol.V3_1_1) {
qos[0] = SUBSCRIBE_ERROR;
} else {
qos[0] = (byte) qoS.ordinal();
}
} else {
qos[0] = (byte) qoS.ordinal();
}

24
activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTAuthTest.java
View File

@ -138,8 +138,9 @@ public class MQTTAuthTest extends MQTTAuthTestSupport {
MQTT mqtt = createMQTTConnection();
mqtt.setClientId("foo");
mqtt.setKeepAlive((short) 2);
mqtt.setVersion("3.1.1");
final BlockingConnection connection = mqtt.blockingConnection();
BlockingConnection connection = mqtt.blockingConnection();
connection.connect();
final String NAMED = "named";
@ -163,7 +164,28 @@ public class MQTTAuthTest extends MQTTAuthTestSupport {
assertEquals(ANONYMOUS, new String(msg.getPayload()));
msg.ack();
//delete retained message
connection.publish(ANONYMOUS, "".getBytes(), QoS.AT_MOST_ONCE, true);
connection.disconnect();
// Test 3.1 functionality
mqtt.setVersion("3.1");
connection = mqtt.blockingConnection();
connection.connect();
qos = connection.subscribe(new Topic[] { new Topic(NAMED, QoS.AT_MOST_ONCE) });
assertEquals(QoS.AT_MOST_ONCE.ordinal(), qos[0]);
MQTT mqttPub = createMQTTConnection("pub", true);
mqttPub.setUserName("admin");
mqttPub.setPassword("admin");
BlockingConnection connectionPub = mqttPub.blockingConnection();
connectionPub.connect();
connectionPub.publish(NAMED, NAMED.getBytes(), QoS.AT_MOST_ONCE, true);
msg = connection.receive(1000, TimeUnit.MILLISECONDS);
assertNull(msg);
}
@Test(timeout = 60 * 1000)