[rabbitmq-discuss] using rabbitmq to rendevous unrelated processes

Fri Feb 1 13:14:16 GMT 2013

On Jan 31, 2013, at 5:09 PM, E R <pc88mxer at gmail.com> wrote:

> I'm wondering if I can use rabbitmq to solve this problem...
> 
> A web client makes a web request which initiates a background server job to perform some work. The client wants to get the result of the server job, but the initial connection might timeout or get broken in which case the web client will re-connect to wait for the result. I want the server job to be able to put the result of the job some place where it can be picked up by the web client at its convenience, and I don't want the web client to have to busy wait for the result.
> 
> Note that is it possible that the web client never picks up the result as well as the job never creating a result, so I need a way to release any created resources after a certain time limit (ie. the client knows that if it doesn't get a result after X minutes it should make a fresh request.)
> 

Hey E,

If you're using Python there's a great plug-in for Celery called Jobtastic:
http://policystat.github.com/jobtastic/

Even if you're not using Python or Celery you could check it out for inspiration.

You *can* accomplish this by using one queue for every job, and declare the queues with the x-expires argument so that if no one is trying to retrieve the result, the queues are removed after some time.

This can be a problem if you have millions (or even hundreds of thousands) of jobs at the same time,
as it means there'll be equally many queues.

Currently I think the best solution here is to store the result in the database.
If you don't want to poll you can create a single 'notification' queue for anyone waiting for a result
(pseudocode below), but chances are you will want to use async javascript on the web to
poll in the background anyway:

An implementation could be:

def wait_for(job_id):
    exchange_name = 'result_notify'
    exchange_declare(exchange_name, type='direct')
    my_notifiy_queue = 'result_notify.%s' % (process_id, ))
    queue_declare(my_notify_queue), auto_delete=True)
    queue_bind(my_notify_queue, exchange_name, str(job_id))

    if result_in_database(job_id):
       return retrieve_result(job_id)
    else:
       basic_consume(my_notify_queue, callback)
       if wait_for_notification(job_id):
           return retrieve_result(job_id)