CODEWITHSUNDEEP
pythonqueuerabbitmqamqppika
Vor
Vor

Reputation: 35149

How to subscribe consumer to N number of unknown queues?

I'm trying to understand how to do this:

Some producers creates N number of queues ( lets say foo.1 foo.2 foo.3 ). Than I have one consumer on another part of the Rabbit that needs to get messages from all N (in my example 3 ) queues. I know I can do something like this:

(method_frame, header_frame, body) = self.channel.basic_get(queue='foo.1', no_ack=False)
(method_frame, header_frame, body) = self.channel.basic_get(queue='foo.2', no_ack=False)
(method_frame, header_frame, body) = self.channel.basic_get(queue='foo.3', no_ack=False)

But what if my consumer doesn't know the names, what I would really like to do is something like:

(method_frame, header_frame, body) = self.channel.basic_get(queue='foo.*', no_ack=False)

Upvotes: 0

Views: 150

Answers (3)

Philip Cristiano
Philip Cristiano

Reputation: 914

The RabbitMQ admin interface/api will have access to all of the queues on the server. There is an easy to use Python client PyRabbit that will let you get_queues. From there you can do whatever filtering you require.

Upvotes: 1

FastTurtle
FastTurtle

Reputation: 2311

If your consumer has a way of recognizing a queue you should be able to find them all by searching through foo.__dict__.

You should keep in mind that if any of your queues are set at the class level, then they will not appear in foo.__dict__. In this case you will have to write an algorithm that traverses foo's mro.

Alternatively if you can modify the creation of queues you can track them through the use of a sort of manager.

class MyQueue(list):
    queues = {}  # Keeps track of all the queues out there

    @classmethod
    def add_to_producer(cls, obj, name, init_values):
        q = MyQueue(init_values)
        cls.queues[(obj, name)] = q
        setattr(obj, name, q)


class MyProducer(object):

    def __init__(self):
        # Initialize our producer with a couple of queues
        MyQueue.add_to_producer(self, 'a', [1,2])
        MyQueue.add_to_producer(self, 'b', [])


p1 = MyProducer()
p2 = MyProducer()
# Add another queue to p2
MyQueue.add_to_producer(p2, 'c', [4,5,6])

# Go through all of our created queues
for obj, attr_name in MyQueue.queues:
    if obj == p1:
        print 'p1', getattr(obj, attr_name)

    if obj == p2:
        print 'p2', getattr(obj, attr_name)

>>> p1 [1, 2]
>>> p1 []
>>> p2 [4, 5, 6]
>>> p2 [1, 2]
>>> p2 []

Upvotes: 1

Sonicjumper
Sonicjumper

Reputation: 31

Here's what I see: You need to get all the messages from a provided number n of queues. From my personal experience, I would just write a for loop and create a string by adding "foo.%s" % (iteration).

Here is an example to what I mean:

for i in range(queues):
    str = 'foo.%s' % (i)
    (method_frame, header_frame, body) = self.channel.basic_get(queue=str, no_ack=False)

As long as you know the number of queues, then you can use this.

Upvotes: 1

Related Questions