I need to setup multiple queues on an exchange. I would like to create a single connection, then declare multiple queues (this works) then, publish messages on the multiple queues (this does not work).
I setup some test code to do this, but it gets hung up on the 2nd publish everytime. I think it does not like publishing on multiple queues without closing the connection, as this code works when I publish on a single queue (even multiple messages on a single queue).
Is there something I need to add to make this work? I would really like to not have to close the connection between publishes. Also, when I have my consumers up, they do not see anything when I send to basic_publish()'s when sending on multiple queues. I do see messages appear almost instantly when I am publishing on a single queue.
#!/usr/bin/env python
import pika
queue_names = ['1a', '2b', '3c', '4d']
# Variables to hold our connection and channel
connection = None
channel = None
# Called when our connection to RabbitMQ is closed
def on_closed(frame):
global connection
# connection.ioloop is blocking, this will stop and exit the app
connection.ioloop.stop()
def on_connected(connection):
"""
Called when we have connected to RabbitMQ
This creates a channel on the connection
"""
global channel #TODO: Test removing this global call
connection.add_on_close_callback(on_closed)
# Create a channel on our connection passing the on_channel_open callback
connection.channel(on_channel_open)
def on_channel_open(channel_):
"""
Called when channel opened
Declare a queue on the channel
"""
global channel
# Our usable channel has been passed to us, assign it for future use
channel = channel_
# Declare a set of queues on this channel
for queue_name in reversed(queue_names):
channel.queue_declare(queue=queue_name, durable=True,
exclusive=False, auto_delete=False,
callback=on_queue_declared)
#print "done making hash"
def on_queue_declared(frame):
"""
Called when a queue is declared
"""
global channel
print "Sending 'Hello World!' on ", frame.method.queue
# Send a message
channel.basic_publish(exchange='',
routing_key=frame.method.queue,
body='Hello World!')
# Create our connection parameters and connect to RabbitMQ
connection = pika.SelectConnection(pika.ConnectionParameters('localhost'), \
on_connected)
# Start our IO/Event loop
try:
connection.ioloop.start()
except KeyboardInterrupt:
print "interrupt"
# Gracefully close the connection
connection.close()
# Loop until we're fully closed, will stop on its own
#connection.ioloop.start()
My solution to this was to have a variable keep track of whether all my queues were declared or not.
In on_queue_declared(), I would check this variable and, if all my queues were declared, then I start publishing messages. I believe attempting to publish messages before getting back all the Queue.DeclareOks was causing the issues.