title: RabbitMQ MQTT Adapter and Paho MQTT client
body: <p>I’m using RabbitMQ MQTT Adapter and Paho MQTT client.</p>
  
  <p>RabbitMQ version: <code>{rabbitmq_mqtt,"RabbitMQ MQTT Adapter","3.2.1"}</code>
  Paho MQTT client version:</p>
  
  <pre><code> &lt;dependency&gt;
          &lt;groupId&gt;org.eclipse.paho&lt;/groupId&gt;
          &lt;artifactId&gt;org.eclipse.paho.client.mqttv3&lt;/artifactId&gt;
          &lt;version&gt;RELEASE&lt;/version&gt;
      &lt;/dependency&gt;
  </code></pre>
  
  <p>Please see code inline.</p>
  
  <p>I’m trying to understand if, the subscriber queue can be durable without expiration time. And If the messages can be durable also.
  As I understood from RabbitMQ documentation, each time a subscriber subscribes to a topic 
  RabbitMQ will create a queue with this naming convention:</p>
  
  <pre><code>mqtt-subscription-&lt;ClientName&gt;qos&lt;ClientQOS&gt;
  </code></pre>
  
  <p>This queue has an expiration time, how can I create a queue without an expiration time? Can I change this queue expiration time to infinite?
  As for now each time I run this command: <code>“service rabbitmq-server restart”</code></p>
  
  <p>The messages in the queue get deleted.</p>
  
  <p>How can I prevent this? Is there a way I can keep the messages in the queue after restart?</p>
  
  <p>In RabbitMQ management UI, I can see under “Publish message” -> “Delivery mode:” which can be “2-persistent”. 
  If I use management UI to publish messages with Delivery mode = 2-persistent. The messages will be in the queue after service restart.</p>
  
  <p>How can I achieve the same using Paho MQTT Client?</p>
  
  <pre><code>// Heavily based on RabbitMQ MQTT adapter test case code!
  
  // first, import the RabbitMQ Java client
  // and the Paho MQTT client classes, plus any other
  // requirements
  
  import com.rabbitmq.client.*;
  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.MqttDeliveryToken;
  import org.eclipse.paho.client.mqttv3.MqttException;
  import org.eclipse.paho.client.mqttv3.MqttMessage;
  import org.eclipse.paho.client.mqttv3.MqttTopic;
  import org.eclipse.paho.client.mqttv3.internal.NetworkModule;
  import org.eclipse.paho.client.mqttv3.internal.TCPNetworkModule;
  // import org.eclipse.paho.client.mqttv3.internal.trace.Trace;
  import org.eclipse.paho.client.mqttv3.internal.wire.MqttOutputStream;
  import org.eclipse.paho.client.mqttv3.internal.wire.MqttPublish;
  
  import javax.net.SocketFactory;
  import java.io.IOException;
  import java.net.InetAddress;
  import java.net.Socket;
  import java.util.*;
  
  /***
   *  MQTT v3.1 tests
   *  TODO: synchronise access to variables
   */
  
  public class MqttTestClean implements MqttCallback {
  
      // setup some variables which define where the MQTT broker is
      private final String host = "0.0.0.0";
      private final int port = 1883;
      private final String brokerUrl = "tcp://" + host + ":" + port;
      private String clientId;
      private String clientId3;
      private MqttClient client;
      private MqttClient client3;
      private MqttConnectOptions conOpt;
      private ArrayList&lt;MqttMessage&gt; receivedMessages;
  
      // specify a message payload - doesn't matter what this says, but since MQTT expects a byte array
      // we convert it from string to byte array here
      private final byte[] payload = "This payload was published on MQTT and read using AMQP.".getBytes();
  
      // specify the topic to be used
      private final String topic = "topic/proxy/1.0.0/Report/*";
  
      private int testDelay = 2000;
      private long lastReceipt;
      private boolean expectConnectionFailure;
  
      private ConnectionFactory connectionFactory;
      private Connection conn;
      private Channel ch;
  
      // override 10s limit
      private class MyConnOpts extends MqttConnectOptions {
          private int keepAliveInterval = 60;
          @Override
          public void setKeepAliveInterval(int keepAliveInterval) {
              this.keepAliveInterval = keepAliveInterval;
          }
          @Override
          public int getKeepAliveInterval() {
              return keepAliveInterval;
          }
      }
  
      public void setUpMqtt() throws MqttException {
          clientId = getClass().getSimpleName() + ((int) (10000*Math.random()));
          client = new MqttClient(brokerUrl, clientId);
          conOpt = new MyConnOpts();
          setConOpts(conOpt);
          receivedMessages = new ArrayList&lt;MqttMessage&gt;();
          expectConnectionFailure = false;
      }
  
      public  void tearDownMqtt() throws MqttException {
          try {
              client.disconnect();
          } catch (Exception _) {}
      }
  
      private void setUpAmqp() throws Exception {
          connectionFactory = new ConnectionFactory();
          connectionFactory.setHost(host);
          conn = connectionFactory.newConnection();
          ch = conn.createChannel();
      }
  
      private void tearDownAmqp() throws IOException {
          conn.close();
      }
  
      private void setConOpts(MqttConnectOptions conOpts) {
          conOpts.setCleanSession(true);
          conOpts.setKeepAliveInterval(60);
      }
  
      private void publish(MqttClient client, String topicName, int qos, byte[] payload) throws MqttException {
          MqttTopic topic = client.getTopic(topicName);
          MqttMessage message = new MqttMessage(payload);
          message.setQos(qos);
          MqttDeliveryToken token = topic.publish(message);
          token.waitForCompletion();
      }
  
      public void connectionLost(Throwable cause) {
          if (!expectConnectionFailure)
              System.out.println("Connection unexpectedly lost");
      }
  
      public void messageArrived(String topic, MqttMessage message) throws Exception {
          lastReceipt = System.currentTimeMillis();
          System.out.println("-------------------------------------------------");
          System.out.println("------------------"  + lastReceipt + "-------------------------------");
          System.out.println("------------------"  + message.toString() + "-------------------------------");
          receivedMessages.add(message);
      }
  
      public void deliveryComplete(IMqttDeliveryToken token) {
      }
  
      public void run() {
          try {
  
              setUpMqtt(); // initialise the MQTT connection
              setUpAmqp(); // initialise the AMQP connection
  
              connect();
  
              //String queue = ch.queueDeclare().getQueue();
              // String queue = ch.queueDeclare("mqtt-subscription-Snabel-3qos1", true, false, false, null).getQueue();
              //ch.queueBind(queue, "amq.topic", "sci-topic.sc.proxy_1393.1.0.0.ApReport.*"/*topic*/);
  
              client.connect(conOpt);
              publish(client, "topic/proxy/1.0.0/Report/123456789",1, payload); // publish the MQTT message
              client.disconnect();
              Thread.sleep(testDelay);
  
              tearDownAmqp(); // cleanup AMQP resources
              tearDownMqtt(); // cleanup MQTT resources*/
              disConnect();
  
          } catch (Exception mqe) {
              mqe.printStackTrace();
          }
      }
  
  
      private void connect() throws Exception {
          clientId3 = "Test-3";
          client3 = new MqttClient(brokerUrl, clientId3);
          MqttConnectOptions connOpts = new MqttConnectOptions();
          connOpts.setCleanSession(false);
  
          client3.connect(connOpts);
          client3.setCallback(this);
          client3.subscribe(topic);
  
          if(!client3.isConnected()){
  
              System.out.println("Not Connected");
              return;
          }
          System.out.println("Connected");
      }
  
      private void disConnect() throws Exception {
          try {
              client3.disconnect();
          } catch (Exception _) {}
      }
  
      public static void main(String[] args) {
          MqttTest mqt = new MqttTest();
          mqt.run();
      }
  }
  </code></pre>