Merge pull request #788 from friedcell/master

Better support for async job payloads

rq/job.py

@@ -552,11 +552,14 @@ class Job(object):
         self.ttl = -1
         _job_stack.push(self.id)
         try:
-            self._result = self.func(*self.args, **self.kwargs)
+            self._result = self._execute()
         finally:
             assert self.id == _job_stack.pop()
         return self._result
 
+    def _execute(self):
+        return self.func(*self.args, **self.kwargs)
+
     def get_ttl(self, default_ttl=None):
         """Returns ttl for a job that determines how long a job will be
         persisted. In the future, this method will also be responsible

rq/queue.py

@@ -232,13 +232,17 @@ class Queue(object):
         job = self.enqueue_job(job, at_front=at_front)
 
         if not self._async:
-            job.perform()
-            job.set_status(JobStatus.FINISHED)
-            job.save(include_meta=False)
-            job.cleanup(DEFAULT_RESULT_TTL)
+            job = self.run_job(job)
 
         return job
 
+    def run_job(self, job):
+        job.perform()
+        job.set_status(JobStatus.FINISHED)
+        job.save(include_meta=False)
+        job.cleanup(DEFAULT_RESULT_TTL)
+        return job
+
     def enqueue(self, f, *args, **kwargs):
         """Creates a job to represent the delayed function call and enqueues
         it.