How subscribe in an RabbitMQ queue with MQTT Paho

5
votes

I am trying to connect from my Android app to one queue called "messages".

The producer (one webservices under AMQP protocol) is already connected, it can be check through RabbitMQ admin panel.

To connect from my Android device I am coding like this.

private void connect() throws Exception {

    this.sampleClient = new MqttClient(this.broker, this.clientId);

    MqttConnectOptions connOpts = new MqttConnectOptions();
    connOpts.setUserName("user");
    connOpts.setPassword("user".toCharArray());
    /*connOpts.setConnectionTimeout(60 * 10);
    connOpts.setKeepAliveInterval(60 * 5);*/
    connOpts.setCleanSession(true);

    this.sampleClient.connect(connOpts);

    this.sampleClient.setCallback(this);

    this.sampleClient.subscribe("messages");

    if(!this.sampleClient.isConnected()){

        System.out.println("Not Connected");
        return;
    }

    System.out.println("Connected");
}

I have tried with "amq.topic", "amq.topic.*", "amq.topic.messages", etc... But when I look in the RabbitMQ queue section "messages" is with 0 consumers, and have been set one new queue called "mqtt-subscription-Sampleqos1" automatically.

What's happening? How can I susbscribe to "messages" queue?

2
rabbitmqmqtt

2 Answers

12
votes

There are two important points about this question.

According with the RabbitMQ MQTT documentation: http://www.rabbitmq.com/mqtt.html

Firstly, every queues are bound automatically to amq.topic exchange by the mqtt-plugin.

Secondly, every subscriber has his own queue which look like this, mqtt-subscription-{cliend_id}{qosX} (where X is the qos level of the subscription)

Therefore, producer must to publish the message to "amq.topic" exchange, and "amq.topic.." routing-key, and receiver must to subscribe to "amq.topic.." routing-key.

2
votes

First, make sure MQTT plugin is enabled: rabbitmq-plugins enable rabbitmq_mqtt

From the client side (here is you Android app), you need subscriber to a topic, lets say, topic my/android/app/messages

this.sampleClient.subscribe("my/android/app/messages");

Then, from the server side, because of RabbitMQ's implementation, you need send the message to a special exchange 'amq.topic' with appropriate route key my.android.app.messages (notice the mapping between '/' and '.', MQTT use / and AMQP use .). For example if you publish by pika AMQP Python lib, the code will looks like following:

channel.basic_publish(
    exchange='amq.topic',
    routing_key='my.android.app.messages',
    body='hello world'
)

In your case, you want to receive message from queue "messages", basically there is no way to directly subscriber message from that AMQP queue on your MQTT client. The work around is create a service running on your server side, work as AMQP subscriber, receive message from "messages" queue, and transparent forward message to exchange amq.topic with proper routing key.

Hope my answer helpful.