title: In Pika or RabbitMQ, How do I check if any consumers are currently consuming?
body: <p>I would like to check if a <em>Consumer/Worker</em> is present to consume a <em>Message</em> I am about to send.</p>
  
  <p>If there isn't any <em>Worker</em>, I would start some workers (both consumers and publishers are on a single machine) and then go about publishing <em>Messages</em>.</p>
  
  <p>If there is a function like <code>connection.check_if_has_consumers</code>, I would implement it somewhat like this - </p>
  
  <pre><code>import pika
  import workers
  
  # code for publishing to worker queue
  connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  channel = connection.channel()
  
  # if there are no consumers running (would be nice to have such a function)
  if not connection.check_if_has_consumers(queue="worker_queue", exchange=""):
      # start the workers in other processes, using python's `multiprocessing`
      workers.start_workers()
  
  # now, publish with no fear of your queues getting filled up
  channel.queue_declare(queue="worker_queue", auto_delete=False, durable=True)
  channel.basic_publish(exchange="", routing_key="worker_queue", body="rockin",
                              properties=pika.BasicProperties(delivery_mode=2))
  connection.close()
  </code></pre>
  
  <p>But I am unable to find any function with <code>check_if_has_consumers</code> functionality in <em>pika</em>.</p>
  
  <p>Is there some way of accomplishing this, using <em>pika</em>? or maybe, by <em>talking</em> to <strong><em>The Rabbit</em></strong> directly?</p>
  
  <p>I am not completely sure, but I really think <em>RabbitMQ</em> would be aware of the number of consumers subscribed to different queues, since it does dispatch <em>messages</em> to them and accepts <em>acks</em></p>
  
  <p>I just got started with <em>RabbitMQ</em> 3 hours ago... any help is welcome...</p>
  
  <p>here is the <strong>workers.py</strong> code I wrote, if its any help....</p>
  
  <pre><code>import multiprocessing
  import pika
  
  
  def start_workers(num=3):
      """start workers as non-daemon processes"""
      for i in xrange(num):    
          process = WorkerProcess()
          process.start()
  
  
  class WorkerProcess(multiprocessing.Process):
      """
      worker process that waits infinitly for task msgs and calls
      the `callback` whenever it gets a msg
      """
      def __init__(self):
          multiprocessing.Process.__init__(self)
          self.stop_working = multiprocessing.Event()
  
      def run(self):
          """
          worker method, open a channel through a pika connection and
          start consuming
          """
          connection = pika.BlockingConnection(
                                pika.ConnectionParameters(host='localhost')
                       )
          channel = connection.channel()
          channel.queue_declare(queue='worker_queue', auto_delete=False,
                                                      durable=True)
  
          # don't give work to one worker guy until he's finished
          channel.basic_qos(prefetch_count=1)
          channel.basic_consume(callback, queue='worker_queue')
  
          # do what `channel.start_consuming()` does but with stopping signal
          while len(channel._consumers) and not self.stop_working.is_set():
              channel.transport.connection.process_data_events()
  
          channel.stop_consuming()
          connection.close()
          return 0
  
      def signal_exit(self):
          """exit when finished with current loop"""
          self.stop_working.set()
  
      def exit(self):
          """exit worker, blocks until worker is finished and dead"""
          self.signal_exit()
          while self.is_alive(): # checking `is_alive()` on zombies kills them
              time.sleep(1)
  
      def kill(self):
          """kill now! should not use this, might create problems"""
          self.terminate()
          self.join()
  
  
  def callback(channel, method, properties, body):
      """pika basic consume callback"""
      print 'GOT:', body
      # do some heavy lifting here
      result = save_to_database(body)
      print 'DONE:', result
      channel.basic_ack(delivery_tag=method.delivery_tag)
  </code></pre>
  
  <p><strong>EDIT:</strong></p>
  
  <p>I have to move forward so here is a workaround that I am going to take, unless a better approach comes along,</p>
  
  <p>So, <em>RabbitMQ</em> has these <a href="http://hg.rabbitmq.com/rabbitmq-management/raw-file/rabbitmq_v2_8_7/priv/www/api/index.html" rel="noreferrer">HTTP management apis</a>, they work after you have turned on the <a href="http://www.rabbitmq.com/management.html" rel="noreferrer">management plugin</a> and at middle of HTTP apis page there is </p>
  
  <blockquote>
    <p>/api/connections - A list of all open connections.</p>
    
    <p>/api/connections/name - An individual connection. DELETEing it will close the connection.</p>
  </blockquote>
  
  <p>So, if I connect my <em>Workers</em> and my <em>Produces</em> both by different <em>Connection</em> names / users, I'll be able to check if the <em>Worker Connection</em> is open... (there might be issues when worker dies...)</p>
  
  <p>will be waiting for a better solution...</p>
  
  <p><strong>EDIT:</strong></p>
  
  <p>just found this in the rabbitmq docs, but this would be hacky to do in python:</p>
  
  <pre><code>shobhit@oracle:~$ sudo rabbitmqctl -p vhostname list_queues name consumers
  Listing queues ...
  worker_queue    0
  ...done.
  </code></pre>
  
  <p>so i could do something like, </p>
  
  <pre><code>subprocess.call("echo password|sudo -S rabbitmqctl -p vhostname list_queues name consumers | grep 'worker_queue'")
  </code></pre>
  
  <p>hacky... still hope pika has some python function to do this...</p>
  
  <p>Thanks,</p>