For AMQ-4338: Lets see if using a newer MQTT client helps. Also hook into the client tracing and use the auto fail test support base class to timeout the tests if they take too long.

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1481092 13f79535-47bb-0310-9956-ffa450edef68

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/AbstractMQTTTest.java

@@ -34,6 +34,7 @@ import javax.jms.TextMessage;
 
 import org.apache.activemq.ActiveMQConnection;
 import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.AutoFailTestSupport;
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.TransportConnector;
 import org.apache.activemq.command.ActiveMQMessage;
@@ -42,7 +43,7 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
-public abstract class AbstractMQTTTest {
+public abstract class AbstractMQTTTest extends AutoFailTestSupport {
     protected TransportConnector mqttConnector;
 
     public static final int AT_MOST_ONCE =0;
@@ -59,7 +60,8 @@ public abstract class AbstractMQTTTest {
     protected int numberOfMessages;
 
     @Before
-    public void startBroker() throws Exception {
+    public void setUp() throws Exception {
+        super.setUp();
         exceptions.clear();
         brokerService = new BrokerService();
         brokerService.setPersistent(false);
@@ -69,10 +71,11 @@ public abstract class AbstractMQTTTest {
     }
 
     @After
-    public void stopBroker() throws Exception {
+    public void tearDown() throws Exception {
         if (brokerService != null) {
             brokerService.stop();
         }
+        super.tearDown();
     }
 
     @Test(timeout=300000)

activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTSSLTest.java

@@ -28,7 +28,7 @@ import org.fusesource.mqtt.client.MQTT;
 
 public class MQTTSSLTest extends MQTTTest {
 
-    public void startBroker() throws Exception {
+    public void setUp() throws Exception {
         String basedir = basedir().getPath();
         System.setProperty("javax.net.ssl.trustStore", basedir+"/src/test/resources/client.keystore");
         System.setProperty("javax.net.ssl.trustStorePassword", "password");
@@ -36,7 +36,7 @@ public class MQTTSSLTest extends MQTTTest {
         System.setProperty("javax.net.ssl.keyStore", basedir+"/src/test/resources/server.keystore");
         System.setProperty("javax.net.ssl.keyStorePassword", "password");
         System.setProperty("javax.net.ssl.keyStoreType", "jks");
-        super.startBroker();
+        super.setUp();
     }
 
     @Override
@@ -46,6 +46,8 @@ public class MQTTSSLTest extends MQTTTest {
 
     protected MQTT createMQTTConnection() throws Exception {
         MQTT mqtt = new MQTT();
+        mqtt.setConnectAttemptsMax(1);
+        mqtt.setTracer(createTracer());
         mqtt.setHost("ssl://localhost:"+mqttConnector.getConnectUri().getPort());
         SSLContext ctx = SSLContext.getInstance("TLS");
         ctx.init(new KeyManager[0], new TrustManager[]{new DefaultTrustManager()}, new SecureRandom());

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

@@ -16,15 +16,19 @@
  */
 package org.apache.activemq.transport.mqtt;
 
-import static org.junit.Assert.assertTrue;
-
 import org.apache.activemq.util.Wait;
 import org.fusesource.mqtt.client.BlockingConnection;
 import org.fusesource.mqtt.client.MQTT;
+import org.fusesource.mqtt.client.Tracer;
+import org.fusesource.mqtt.codec.MQTTFrame;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class MQTTTest extends AbstractMQTTTest {
 
+    private static final Logger LOG = LoggerFactory.getLogger(MQTTTest.class);
+
     @Test(timeout=300000)
     public void testPingKeepsInactivityMonitorAlive() throws Exception {
         addMQTTConnector();
@@ -128,10 +132,32 @@ public class MQTTTest extends AbstractMQTTTest {
 
     protected MQTT createMQTTConnection() throws Exception {
         MQTT mqtt = new MQTT();
+        mqtt.setConnectAttemptsMax(1);
+        mqtt.setTracer(createTracer());
         mqtt.setHost("localhost", mqttConnector.getConnectUri().getPort());
         // shut off connect retry
         mqtt.setConnectAttemptsMax(0);
         mqtt.setReconnectAttemptsMax(0);
         return mqtt;
     }
+
+    protected Tracer createTracer() {
+        return new Tracer(){
+            @Override
+            public void onReceive(MQTTFrame frame) {
+                LOG.info("recv: "+frame);
+            }
+
+            @Override
+            public void onSend(MQTTFrame frame) {
+                LOG.info("send: " + frame);
+            }
+
+            @Override
+            public void debug(String message, Object... args) {
+                LOG.info(message, args);
+            }
+        };
+    }
+
 }

pom.xml

@@ -88,7 +88,7 @@
     <leveldb-version>0.5</leveldb-version>
     <leveldbjni-version>1.6</leveldbjni-version>
     <log4j-version>1.2.17</log4j-version>
-    <mqtt-client-version>1.4</mqtt-client-version>
+    <mqtt-client-version>1.5-SNAPSHOT</mqtt-client-version>
     <openjpa-version>1.2.0</openjpa-version>
     <opensymphony-version>2.4.2</opensymphony-version>
     <org-apache-derby-version>10.9.1.0</org-apache-derby-version>