switch to our custom Task class, handle rabbit downtime (bug 583314)

1 files changed, 28 insertions, 0 deletions

diff --git a/lib/celeryutils.py b/lib/celeryutils.py
new file mode 100644
index 0000000..ee81aa1
--- /dev/null
+++ b/lib/celeryutils.py
@@ -0,0 +1,28 @@
+import logging
+import functools
+
+import celery.decorators
+import celery.task
+
+
+log = logging.getLogger('z.celery')
+
+
+class Task(celery.task.Task):
+
+    @classmethod
+    def apply_async(self, args=None, kwargs=None, **options):
+        try:
+            return super(Task, self).apply_async(args, kwargs, **options)
+        except Exception, e:
+            log.error('CELERY FAIL: %s' % e)
+
+
+def task(*args, **kw):
+    # Force usage of our Task subclass.
+    kw['base'] = Task
+    wrapper = celery.decorators.task(**kw)
+    if args:
+        return wrapper(*args)
+    else:
+        return wrapper