Merge branch 'KanbanSolutions-use-pipeline-context'

main
Vincent Driessen 11 years ago
commit 9ad7da684c

@ -416,6 +416,7 @@ class Worker(object):
job.func_name, job.func_name,
job.origin, time.time())) job.origin, time.time()))
with self.connection._pipeline() as pipeline:
try: try:
with death_penalty_after(job.timeout or Queue.DEFAULT_TIMEOUT): with death_penalty_after(job.timeout or Queue.DEFAULT_TIMEOUT):
rv = job.perform() rv = job.perform()
@ -427,13 +428,12 @@ class Worker(object):
job.ended_at = utcnow() job.ended_at = utcnow()
result_ttl = job.get_ttl(self.default_result_ttl) result_ttl = job.get_ttl(self.default_result_ttl)
pipeline = self.connection._pipeline()
if result_ttl != 0: if result_ttl != 0:
job.save(pipeline=pipeline) job.save(pipeline=pipeline)
job.cleanup(result_ttl, pipeline=pipeline) job.cleanup(result_ttl, pipeline=pipeline)
pipeline.execute() pipeline.execute()
except: except Exception:
# Use the public setter here, to immediately update Redis # Use the public setter here, to immediately update Redis
job.status = Status.FAILED job.status = Status.FAILED
self.handle_exception(job, *sys.exc_info()) self.handle_exception(job, *sys.exc_info())

Loading…
Cancel
Save