Merge branch 'sdcooke/task_expires'

celery/execute/__init__.py

@@ -96,7 +96,10 @@ def apply_async(task, args=None, kwargs=None, countdown=None, eta=None,
     options = router.route(options, task.name, args, kwargs)
     exchange = options.get("exchange")
     exchange_type = options.get("exchange_type")
-
+    
+    if not expires: # If expires has not been passed to this function, get it from the task (it may still be None)
+        expires = task.expires
+    
     publish = publisher or task.get_publisher(connection, exchange=exchange,
                                               exchange_type=exchange_type)
     try:

celery/task/base.py

@@ -235,6 +235,7 @@ class Task(object):
     disable_error_emails = False                                    # FIXME
     max_retries = 5
     default_retry_delay = 3 * 60
+    expires = None
     serializer = conf.TASK_SERIALIZER
     rate_limit = conf.DEFAULT_RATE_LIMIT
     backend = default_backend