app.py

#!.ve/bin/python
import gc
import os
os.environ['MP_LOG'] = '1'
os.environ['CELERY_RDBSIG'] = '1'
import time
import sys
from celery import Celery

celery = Celery(
    broker="amqp://test:test@localhost:5672/test"
)
celery.conf.update(
    CELERY_TASK_SERIALIZER = "pickle",
    CELERYD_MAX_TASKS_PER_CHILD = 1,
    #CELERY_WORKER_DIRECT = True,
    #BROKER_POOL_LIMIT = None,
    CELERY_RESULT_BACKEND = "mongodb",
    CELERY_MONGODB_BACKEND_SETTINGS = {},
    #CELERY_DISABLE_RATE_LIMITS = True,
    #CELERYD_PREFETCH_MULTIPLIER = 0,
    #CELERY_MAX_CACHED_RESULTS = 0,
)

@celery.task
def stuff():
    return "ok"

if __name__ == '__main__':
    if len(sys.argv) > 1 and sys.argv[1] == 'run-test':
        while 1:
            try:
                async_results = [stuff.apply_async(
                    args=(),
                    options={
                        'queue_name': None,
                        'exchange': 'C.dq',
                        'routing_key': file('/etc/hostname').read().strip(),
                    }
                ) for i in range(10)]
                results = [res.get(propagate=False) for res in async_results]
                assert results == ['ok' for i in range(10)], "BAD RESULTS: %r" % results
                print "OK."
            except Exception:
                import traceback
                traceback.print_exc()
                print 'wait 2 secs ...'
            time.sleep(0.5)
    else:
        celery.start()

bootstrap.sh

#!/bin/bash -eEx

rm -rf .ve
virtualenv .ve

.ve/bin/pip install -I -U --download-cache=.ve/.cache \
    https://github.com/celery/py-amqp/archive/master.zip \
    https://github.com/celery/billiard/archive/master.zip \
    https://github.com/celery/kombu/archive/master.zip \
    pymongo \
    https://github.com/celery/celery/archive/master.zip \
    objgraph ipython


sudo rabbitmqctl delete_vhost test
sudo rabbitmqctl add_vhost test
sudo rabbitmqctl add_user test test || echo test user already created
sudo rabbitmqctl set_permissions -p test test ".*" ".*" ".*"

INSTRUCTIONS.rst

Steps to reproduce

In first terminal

./bootstrap.sh
./app.py worker --events --purge --loglevel=DEBUG --concurrency 5

In second terminal

./app.py run-test