RabbitMQ - pika - rabbit.js - node.js

Question

I'm using Python, rabbitmq, pika, rabbit.js and node.js.

The idea is to send messages to clients and act accordingly. There are multiple types of messages being sent.

So, on the server side I have a method which will receive the message and the exchange to send the message to:

import pika
def send(exchange, message):
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange=exchange, type='fanout',)
    channel.basic_publish(exchange=exchange,
                          routing_key='',
                          body=message)

Still on the server side, using node.js, I have the multiple contexts, which, I believe, the exchanges will call according to the name: (Note: "clients" refers to the list of clients connected to the server.)

var context = require('rabbit.js').createContext();

context.on('ready', function() {

    var consumer_1 = context.socket('SUB');
    var consumer_2 = context.socket('SUB');

    consumer_1.setEncoding('utf8');
    consumer_1.connect('consumer_1', function (){
    consumer_1.on('data', function(data){
        try {
            io.sockets.emit("client_method_1", data);
        }catch(err){
            console.log("[CONSUMER 1 ERROR] " + err);
        }

    });

    consumer_2.setEncoding('utf8');
    consumer_2.connect('client_method_2', function (){
    consumer_2.on('data', function(data){
        try {
            clients[data.client].emit('client_method_2', data.action );
        }
        catch(err) {
            console.log("[CONSUMER 2 ERROR]: " + err);
        }
    });

});

What's happening here is that consumer_1 is emitting messages to all clients and consumer_2 is emitting to a specific client and that part is working well.

What's wrong is that all consumers are receiving messages from all exchanges. If I try to send a message to consumer_1, the consumer_2 also receives the message.

On the practical side, if I call:

send('consumer_1', data)

The result would be:

• node will emit to 'client_method_1';
• and also will try to emit to 'client_method_2'.

How do I call a specific consumer without calling every other?

Or, should I declare only one consumer and filter the data to call different methods?

EDIT:

Following TokenMacGuy advice, I've tried to change the exchange type to "topic" or "direct". That should do exactly what I want.

But there's always a but, somehow I can't redeclare the exchange. When I restart node I keep getting this error:

[(406, "PRECONDITION_FAILED - cannot redeclare exchange 'consumer_1' in vhost '/' with different type, durable, internal or autodelete value")]

Ok, so I've checked the list_exchanged on rabbitmq and the consumer is still with "fanout".

    direct
amq.direct  direct
amq.fanout  fanout
amq.headers headers
amq.match   headers
amq.rabbitmq.log    topic
amq.rabbitmq.trace  topic
amq.topic   topic
celery  direct
celery.pidbox   fanout
consumer_1    fanout
consumer_2    fanout
consumer_3    fanout

Next move was to reset the rabbitmq by stoping, reseting (and later force reseting) and starting it.

I've send a message to consumer_1

Now the consumer_1 is listed as "direct":

    direct
amq.direct  direct
amq.fanout  fanout
amq.headers headers
amq.match   headers
amq.rabbitmq.log    topic
amq.rabbitmq.trace  topic
amq.topic   topic
celery  direct
celery.pidbox   fanout
consumer_1    direct
consumer_2    fanout
consumer_3    fanout

Ok, since I've only called consumer_1 and all other appear on the list there must be a file with that information. Where? I have no idea!!

Anyway, same PRECONDITION_FAILED error occurs on consumer_1.

Any ideas?

Answer 1

The fanout exchange type has only one behavior, delivering messages to every queue bound to it. The other exchange types can deliver messages with more narrow criteria. The most convenient is probably the topic exchange type, which takes a binding key consisting of wild cards, and causes all messages which match that wild card to be delivered to those specific queues. (here's a link that goes into some detail: http://www.rabbitmq.com/tutorials/tutorial-five-python.html)