From 01ab2f20dd90db569ec7ea954fb0c3341fee88d2 Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Fri, 23 Jan 2015 13:58:25 +0700
Subject: [PATCH 01/16] Minor copywriting fixes.

---
 rq/worker.py | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/rq/worker.py b/rq/worker.py
index 3ba297a..c0cb296 100644
--- a/rq/worker.py
+++ b/rq/worker.py
@@ -67,10 +67,13 @@ def signal_name(signum):
         return 'SIG_UNKNOWN'
 
 
-WorkerStatus = enum('WorkerStatus',
-                    STARTED='started', SUSPENDED='suspended', BUSY='busy',
-                    IDLE='idle'
-                    )
+WorkerStatus = enum(
+    'WorkerStatus',
+    STARTED='started',
+    SUSPENDED='suspended',
+    BUSY='busy',
+    IDLE='idle'
+)
 
 
 class Worker(object):
@@ -342,7 +345,7 @@ class Worker(object):
         signal.signal(signal.SIGTERM, request_stop)
 
     def check_for_suspension(self, burst):
-        """Check to see if the workers have been suspended by something like `rq suspend`"""
+        """Check to see if workers have been suspended by `rq suspend`"""
 
         before_state = None
         notified = False
@@ -350,8 +353,8 @@ class Worker(object):
         while not self.stopped and is_suspended(self.connection):
 
             if burst:
-                self.log.info('Suspended in burst mode -- exiting.')
-                self.log.info('Note:  There could still be unperformed jobs on the queue')
+                self.log.info('Suspended in burst mode -- exiting.'
+                              'Note: There could still be unperformed jobs on the queue')
                 raise StopRequested
 
             if not notified:

From dc5cd514ee757b401415bd1dccdf70831c4797ef Mon Sep 17 00:00:00 2001
From: Vincent Driessen <vincent@3rdcloud.com>
Date: Wed, 21 Jan 2015 16:12:08 -0800
Subject: [PATCH 02/16] Explicitly cast map() result to a list, for Python 3
 compat.

---
 rq/worker.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/rq/worker.py b/rq/worker.py
index c0cb296..b971a24 100644
--- a/rq/worker.py
+++ b/rq/worker.py
@@ -169,11 +169,11 @@ class Worker(object):
 
     def queue_names(self):
         """Returns the queue names of this worker's queues."""
-        return map(lambda q: q.name, self.queues)
+        return list(map(lambda q: q.name, self.queues))
 
     def queue_keys(self):
         """Returns the Redis keys representing this worker's queues."""
-        return map(lambda q: q.key, self.queues)
+        return list(map(lambda q: q.key, self.queues))
 
     @property
     def name(self):

From 7fd2ac8ca6bc737974d11a6f2aedd20679a5fa17 Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Fri, 23 Jan 2015 15:16:32 +0700
Subject: [PATCH 03/16] Added "DEFERRED" Job status for jobs that have
 unsatisfied dependencies.

---
 rq/job.py           | 11 ++++++++---
 rq/queue.py         |  2 +-
 tests/test_queue.py |  6 ++++--
 3 files changed, 13 insertions(+), 6 deletions(-)

diff --git a/rq/job.py b/rq/job.py
index 9afe578..2337ad4 100644
--- a/rq/job.py
+++ b/rq/job.py
@@ -25,9 +25,14 @@ dumps = partial(pickle.dumps, protocol=pickle.HIGHEST_PROTOCOL)
 loads = pickle.loads
 
 
-JobStatus = enum('JobStatus',
-              QUEUED='queued', FINISHED='finished', FAILED='failed',
-              STARTED='started')
+JobStatus = enum(
+    'JobStatus',
+    QUEUED='queued',
+    FINISHED='finished',
+    FAILED='failed',
+    STARTED='started',
+    DEFERRED='deferred'
+)
 
 # Sentinel value to mark that some of our lazily evaluated properties have not
 # yet been evaluated.
diff --git a/rq/queue.py b/rq/queue.py
index 7352459..9cc0fa0 100644
--- a/rq/queue.py
+++ b/rq/queue.py
@@ -182,7 +182,6 @@ class Queue(object):
         """
         timeout = timeout or self._default_timeout
 
-        # TODO: job with dependency shouldn't have "queued" as status
         job = self.job_class.create(func, args, kwargs, connection=self.connection,
                                     result_ttl=result_ttl, status=JobStatus.QUEUED,
                                     description=description, depends_on=depends_on, timeout=timeout,
@@ -200,6 +199,7 @@ class Queue(object):
                     try:
                         pipe.watch(depends_on.key)
                         if depends_on.get_status() != JobStatus.FINISHED:
+                            job.set_status(JobStatus.DEFERRED)
                             job.register_dependency(pipeline=pipe)
                             job.save(pipeline=pipe)
                             pipe.execute()
diff --git a/tests/test_queue.py b/tests/test_queue.py
index 0a84e78..cddb3f6 100644
--- a/tests/test_queue.py
+++ b/tests/test_queue.py
@@ -342,8 +342,9 @@ class TestQueue(RQTestCase):
         # Job with unfinished dependency is not immediately enqueued
         parent_job = Job.create(func=say_hello)
         q = Queue()
-        q.enqueue_call(say_hello, depends_on=parent_job)
+        job = q.enqueue_call(say_hello, depends_on=parent_job)
         self.assertEqual(q.job_ids, [])
+        self.assertEqual(job.get_status(), JobStatus.DEFERRED)
 
         # Jobs dependent on finished jobs are immediately enqueued
         parent_job.set_status(JobStatus.FINISHED)
@@ -351,6 +352,7 @@ class TestQueue(RQTestCase):
         job = q.enqueue_call(say_hello, depends_on=parent_job)
         self.assertEqual(q.job_ids, [job.id])
         self.assertEqual(job.timeout, Queue.DEFAULT_TIMEOUT)
+        self.assertEqual(job.get_status(), JobStatus.QUEUED)
 
     def test_enqueue_job_with_dependency_by_id(self):
         """Enqueueing jobs should work as expected by id as well as job-objects."""
@@ -368,7 +370,7 @@ class TestQueue(RQTestCase):
         self.assertEqual(job.timeout, Queue.DEFAULT_TIMEOUT)
 
     def test_enqueue_job_with_dependency_and_timeout(self):
-        """Jobs still know their specified timeout after being scheduled as a dependency."""
+        """Jobs remember their timeout when enqueued as a dependency."""
         # Job with unfinished dependency is not immediately enqueued
         parent_job = Job.create(func=say_hello)
         q = Queue()

From 9320496402c1a92d946eaac28bfb03aae194c0b0 Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Fri, 23 Jan 2015 16:05:29 +0700
Subject: [PATCH 04/16] Simplify FailedQueue.quarantine and ensure that a
 deferred job's status is set to Queued when enqueued.

---
 rq/queue.py | 41 ++++++++++++++++++++++++-----------------
 1 file changed, 24 insertions(+), 17 deletions(-)

diff --git a/rq/queue.py b/rq/queue.py
index 9cc0fa0..eb50314 100644
--- a/rq/queue.py
+++ b/rq/queue.py
@@ -248,24 +248,25 @@ class Queue(object):
                                  description=description, depends_on=depends_on,
                                  job_id=job_id, at_front=at_front)
 
-    def enqueue_job(self, job, set_meta_data=True, at_front=False):
+    def enqueue_job(self, job, at_front=False):
         """Enqueues a job for delayed execution.
 
-        If the `set_meta_data` argument is `True` (default), it will update
-        the properties `origin` and `enqueued_at`.
-
         If Queue is instantiated with async=False, job is executed immediately.
         """
-        # Add Queue key set
-        self.connection.sadd(self.redis_queues_keys, self.key)
+        
+        with self.connection._pipeline() as pipeline:
+            # Add Queue key set
+            self.connection.sadd(self.redis_queues_keys, self.key)
+            job.set_status(JobStatus.QUEUED, pipeline=pipeline)
 
-        if set_meta_data:
             job.origin = self.name
             job.enqueued_at = utcnow()
 
-        if job.timeout is None:
-            job.timeout = self.DEFAULT_TIMEOUT
-        job.save()
+            if job.timeout is None:
+                job.timeout = self.DEFAULT_TIMEOUT
+            job.save(pipeline=pipeline)
+
+            pipeline.execute()
 
         if self._async:
             self.push_job_id(job.id, at_front=at_front)
@@ -401,14 +402,20 @@ class FailedQueue(Queue):
     def quarantine(self, job, exc_info):
         """Puts the given Job in quarantine (i.e. put it on the failed
         queue).
-
-        This is different from normal job enqueueing, since certain meta data
-        must not be overridden (e.g. `origin` or `enqueued_at`) and other meta
-        data must be inserted (`ended_at` and `exc_info`).
         """
-        job.ended_at = utcnow()
-        job.exc_info = exc_info
-        return self.enqueue_job(job, set_meta_data=False)
+
+        with self.connection._pipeline() as pipeline:
+            # Add Queue key set
+            self.connection.sadd(self.redis_queues_keys, self.key)
+
+            job.ended_at = utcnow()
+            job.exc_info = exc_info
+            job.save(pipeline=pipeline)
+
+            self.push_job_id(job.id, pipeline=pipeline)
+            pipeline.execute()
+
+        return job
 
     def requeue(self, job_id):
         """Requeues the job with the given job ID."""

From 3e674fbe6aa20162c094e93fb9a01ead2dabc43d Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Fri, 23 Jan 2015 16:33:48 +0700
Subject: [PATCH 05/16] queue.enqueue() should set job.origin.

---
 rq/job.py           | 7 +++++--
 rq/queue.py         | 9 +++++----
 tests/test_queue.py | 3 +--
 3 files changed, 11 insertions(+), 8 deletions(-)

diff --git a/rq/job.py b/rq/job.py
index 2337ad4..fbf7d5a 100644
--- a/rq/job.py
+++ b/rq/job.py
@@ -88,8 +88,8 @@ class Job(object):
     # Job construction
     @classmethod
     def create(cls, func, args=None, kwargs=None, connection=None,
-               result_ttl=None, ttl=None, status=None, description=None, depends_on=None, timeout=None,
-               id=None):
+               result_ttl=None, ttl=None, status=None, description=None,
+               depends_on=None, timeout=None, id=None, origin=None):
         """Creates a new Job instance for the given function, arguments, and
         keyword arguments.
         """
@@ -107,6 +107,9 @@ class Job(object):
         if id is not None:
             job.set_id(id)
 
+        if origin is not None:
+            job.origin = origin
+
         # Set the core job tuple properties
         job._instance = None
         if inspect.ismethod(func):
diff --git a/rq/queue.py b/rq/queue.py
index eb50314..d596106 100644
--- a/rq/queue.py
+++ b/rq/queue.py
@@ -182,10 +182,11 @@ class Queue(object):
         """
         timeout = timeout or self._default_timeout
 
-        job = self.job_class.create(func, args, kwargs, connection=self.connection,
-                                    result_ttl=result_ttl, status=JobStatus.QUEUED,
-                                    description=description, depends_on=depends_on, timeout=timeout,
-                                    id=job_id)
+        job = self.job_class.create(
+            func, args, kwargs, connection=self.connection,
+            result_ttl=result_ttl, status=JobStatus.QUEUED,
+            description=description, depends_on=depends_on,
+            timeout=timeout, id=job_id, origin=self.name)
 
         # If job depends on an unfinished job, register itself on it's
         # parent's dependents instead of enqueueing it.
diff --git a/tests/test_queue.py b/tests/test_queue.py
index cddb3f6..f5edcbf 100644
--- a/tests/test_queue.py
+++ b/tests/test_queue.py
@@ -117,6 +117,7 @@ class TestQueue(RQTestCase):
         # say_hello spec holds which queue this is sent to
         job = q.enqueue(say_hello, 'Nick', foo='bar')
         job_id = job.id
+        self.assertEqual(job.origin, q.name)
 
         # Inspect data inside Redis
         q_key = 'rq:queue:default'
@@ -131,14 +132,12 @@ class TestQueue(RQTestCase):
         job = Job.create(func=say_hello, args=('Nick',), kwargs=dict(foo='bar'))
 
         # Preconditions
-        self.assertIsNone(job.origin)
         self.assertIsNone(job.enqueued_at)
 
         # Action
         q.enqueue_job(job)
 
         # Postconditions
-        self.assertEquals(job.origin, q.name)
         self.assertIsNotNone(job.enqueued_at)
 
     def test_pop_job_id(self):

From dac0be6cc79dd78f8039e4b5b989db425ed380b3 Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Fri, 23 Jan 2015 17:53:27 +0700
Subject: [PATCH 06/16] Added DeferredJobsRegistry to keep track of deferred
 jobs.

---
 rq/job.py              |  8 +++++++-
 rq/queue.py            |  5 +++++
 rq/registry.py         | 18 +++++++++++++++++-
 tests/test_job.py      | 11 +++++++++--
 tests/test_queue.py    | 22 ++++++++++++++--------
 tests/test_registry.py | 19 ++++++++++++++++++-
 6 files changed, 70 insertions(+), 13 deletions(-)

diff --git a/rq/job.py b/rq/job.py
index fbf7d5a..d815c5b 100644
--- a/rq/job.py
+++ b/rq/job.py
@@ -544,8 +544,14 @@ class Job(object):
 
             rq:job:job_id:dependents = {'job_id_1', 'job_id_2'}
 
-        This method adds the current job in its dependency's dependents set.
+        This method adds the job in its dependency's dependents set
+        and adds the job to DeferredJobRegistry.
         """
+        from .registry import DeferredJobRegistry
+
+        registry = DeferredJobRegistry(self.origin, connection=self.connection)
+        registry.add(self, pipeline=pipeline)
+
         connection = pipeline if pipeline is not None else self.connection
         connection.sadd(Job.dependents_key_for(self._dependency_id), self.id)
 
diff --git a/rq/queue.py b/rq/queue.py
index d596106..e2c358c 100644
--- a/rq/queue.py
+++ b/rq/queue.py
@@ -279,11 +279,16 @@ class Queue(object):
     def enqueue_dependents(self, job):
         """Enqueues all jobs in the given job's dependents set and clears it."""
         # TODO: can probably be pipelined
+        from .registry import DeferredJobRegistry
+
+        registry = DeferredJobRegistry(self.name, self.connection)
+
         while True:
             job_id = as_text(self.connection.spop(job.dependents_key))
             if job_id is None:
                 break
             dependent = self.job_class.fetch(job_id, connection=self.connection)
+            registry.remove(dependent)
             self.enqueue_job(dependent)
 
     def pop_job_id(self):
diff --git a/rq/registry.py b/rq/registry.py
index b4cf43f..08798eb 100644
--- a/rq/registry.py
+++ b/rq/registry.py
@@ -24,7 +24,7 @@ class BaseRegistry(object):
         self.cleanup()
         return self.connection.zcard(self.key)
 
-    def add(self, job, ttl, pipeline=None):
+    def add(self, job, ttl=0, pipeline=None):
         """Adds a job to a registry with expiry time of now + ttl."""
         score = ttl if ttl < 0 else current_timestamp() + ttl
         if pipeline is not None:
@@ -108,3 +108,19 @@ class FinishedJobRegistry(BaseRegistry):
         """
         score = timestamp if timestamp is not None else current_timestamp()
         self.connection.zremrangebyscore(self.key, 0, score)
+
+
+class DeferredJobRegistry(BaseRegistry):
+    """
+    Registry of deferred jobs (waiting for another job to finish).
+    """
+
+    def __init__(self, name='default', connection=None):
+        super(DeferredJobRegistry, self).__init__(name, connection)
+        self.key = 'rq:deferred:%s' % name
+
+    def cleanup(self):
+        """This method is only here to prevent errors because this method is
+        automatically called by `count()` and `get_job_ids()` methods
+        implemented in BaseRegistry."""
+        pass
diff --git a/tests/test_job.py b/tests/test_job.py
index 34859a7..b7854b3 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -7,6 +7,7 @@ from datetime import datetime
 from rq.compat import as_text, PY2
 from rq.exceptions import NoSuchJobError, UnpickleError
 from rq.job import get_current_job, Job
+from rq.registry import DeferredJobRegistry
 from rq.queue import Queue
 from rq.utils import utcformat
 
@@ -331,12 +332,18 @@ class TestJob(RQTestCase):
         self.assertRaises(NoSuchJobError, Job.fetch, job.id, self.testconn)
 
     def test_register_dependency(self):
-        """Test that jobs updates the correct job dependents."""
-        job = Job.create(func=say_hello)
+        """Ensure dependency registration works properly."""
+        origin = 'some_queue'
+        registry = DeferredJobRegistry(origin, self.testconn)
+
+        job = Job.create(func=say_hello, origin=origin)
         job._dependency_id = 'id'
         job.save()
+        
+        self.assertEqual(registry.get_job_ids(), [])
         job.register_dependency()
         self.assertEqual(as_text(self.testconn.spop('rq:job:id:dependents')), job.id)
+        self.assertEqual(registry.get_job_ids(), [job.id])
 
     def test_cancel(self):
         """job.cancel() deletes itself & dependents mapping from Redis."""
diff --git a/tests/test_queue.py b/tests/test_queue.py
index f5edcbf..369b3f1 100644
--- a/tests/test_queue.py
+++ b/tests/test_queue.py
@@ -5,6 +5,7 @@ from __future__ import (absolute_import, division, print_function,
 from rq import get_failed_queue, Queue
 from rq.exceptions import InvalidJobOperationError
 from rq.job import Job, JobStatus
+from rq.registry import DeferredJobRegistry
 from rq.worker import Worker
 
 from tests import RQTestCase
@@ -319,23 +320,28 @@ class TestQueue(RQTestCase):
         self.assertEquals(len(Queue.all()), 3)
 
     def test_enqueue_dependents(self):
-        """Enqueueing the dependent jobs pushes all jobs in the depends set to the queue."""
+        """Enqueueing dependent jobs pushes all jobs in the depends set to the queue
+        and removes them from DeferredJobQueue."""
         q = Queue()
         parent_job = Job.create(func=say_hello)
         parent_job.save()
-        job_1 = Job.create(func=say_hello, depends_on=parent_job)
-        job_1.save()
-        job_1.register_dependency()
-        job_2 = Job.create(func=say_hello, depends_on=parent_job)
-        job_2.save()
-        job_2.register_dependency()
+        job_1 = q.enqueue(say_hello, depends_on=parent_job)
+        job_2 = q.enqueue(say_hello, depends_on=parent_job)
 
+        registry = DeferredJobRegistry(q.name, connection=self.testconn)
+        self.assertEqual(
+            set(registry.get_job_ids()),
+            set([job_1.id, job_2.id])
+        )
         # After dependents is enqueued, job_1 and job_2 should be in queue
         self.assertEqual(q.job_ids, [])
         q.enqueue_dependents(parent_job)
-        self.assertEqual(set(q.job_ids), set([job_1.id, job_2.id]))
+        self.assertEqual(set(q.job_ids), set([job_2.id, job_1.id]))
         self.assertFalse(self.testconn.exists(parent_job.dependents_key))
 
+        # DeferredJobRegistry should also be empty
+        self.assertEqual(registry.get_job_ids(), [])        
+
     def test_enqueue_job_with_dependency(self):
         """Jobs are enqueued only when their dependencies are finished."""
         # Job with unfinished dependency is not immediately enqueued
diff --git a/tests/test_registry.py b/tests/test_registry.py
index 26470e3..3798a8f 100644
--- a/tests/test_registry.py
+++ b/tests/test_registry.py
@@ -5,7 +5,8 @@ from rq.job import Job
 from rq.queue import FailedQueue, Queue
 from rq.utils import current_timestamp
 from rq.worker import Worker
-from rq.registry import FinishedJobRegistry, StartedJobRegistry
+from rq.registry import (DeferredJobRegistry, FinishedJobRegistry,
+                         StartedJobRegistry)
 
 from tests import RQTestCase
 from tests.fixtures import div_by_zero, say_hello
@@ -135,3 +136,19 @@ class TestFinishedJobRegistry(RQTestCase):
         failed_job = queue.enqueue(div_by_zero)
         worker.perform_job(failed_job)
         self.assertEqual(self.registry.get_job_ids(), [job.id])
+
+
+class TestRegistry(RQTestCase):
+
+    def setUp(self):
+        super(TestRegistry, self).setUp()
+        self.registry = DeferredJobRegistry(connection=self.testconn)
+
+    def test_add(self):
+        """Adding a job to DeferredJobsRegistry."""
+        job = Job()
+        self.registry.add(job)
+        self.assertEqual(
+            self.testconn.zrange(self.registry.key, 0, -1),
+            [job.id]
+        )
\ No newline at end of file

From de1cd8a83c8f1da732921d7b4536371f8b6e06b9 Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Mon, 26 Jan 2015 08:31:58 +0700
Subject: [PATCH 07/16] Fixed test error in Python 3.

---
 tests/test_registry.py | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/tests/test_registry.py b/tests/test_registry.py
index 3798a8f..5f5cf75 100644
--- a/tests/test_registry.py
+++ b/tests/test_registry.py
@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import
 
+from rq.compat import as_text
 from rq.job import Job
 from rq.queue import FailedQueue, Queue
 from rq.utils import current_timestamp
@@ -120,7 +121,6 @@ class TestFinishedJobRegistry(RQTestCase):
         self.registry.cleanup(timestamp + 20)
         self.assertEqual(self.registry.get_job_ids(), ['baz'])
 
-
     def test_jobs_are_put_in_registry(self):
         """Completed jobs are added to FinishedJobRegistry."""
         self.assertEqual(self.registry.get_job_ids(), [])
@@ -138,7 +138,7 @@ class TestFinishedJobRegistry(RQTestCase):
         self.assertEqual(self.registry.get_job_ids(), [job.id])
 
 
-class TestRegistry(RQTestCase):
+class TestDeferredRegistry(RQTestCase):
 
     def setUp(self):
         super(TestRegistry, self).setUp()
@@ -148,7 +148,6 @@ class TestRegistry(RQTestCase):
         """Adding a job to DeferredJobsRegistry."""
         job = Job()
         self.registry.add(job)
-        self.assertEqual(
-            self.testconn.zrange(self.registry.key, 0, -1),
-            [job.id]
-        )
\ No newline at end of file
+        job_ids = [as_text(job_id) for job_id in
+                   self.testconn.zrange(self.registry.key, 0, -1)]
+        self.assertEqual(job_ids, [job.id])

From 105b95e9b8eaa20bbd9b1a47742ffef71d2bd694 Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Mon, 26 Jan 2015 08:37:03 +0700
Subject: [PATCH 08/16] Fixed an error in super call.

---
 tests/test_registry.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_registry.py b/tests/test_registry.py
index 5f5cf75..f54b315 100644
--- a/tests/test_registry.py
+++ b/tests/test_registry.py
@@ -141,7 +141,7 @@ class TestFinishedJobRegistry(RQTestCase):
 class TestDeferredRegistry(RQTestCase):
 
     def setUp(self):
-        super(TestRegistry, self).setUp()
+        super(TestDeferredRegistry, self).setUp()
         self.registry = DeferredJobRegistry(connection=self.testconn)
 
     def test_add(self):

From 008d72ff09e77e8186b81955ebac9509fd85b701 Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Fri, 30 Jan 2015 14:39:42 +0700
Subject: [PATCH 09/16] Documented new features in 0.5.0.

---
 CHANGES.md | 22 ++++++++++++++++++++++
 1 file changed, 22 insertions(+)

diff --git a/CHANGES.md b/CHANGES.md
index 9097cb9..12f273a 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,3 +1,25 @@
+### 0.5.0
+(Jan 30th, 2015)
+
+- RQ workers can now be paused and resumed using `rq suspend` and
+  `rq resume` commands. Thanks Jonathan Tushman!
+- Jobs that are being performed are now stored in `StartedJobRegistry`
+  for monitoring purposes. This also prevents currently active jobs from
+  being orphaned/lost in the case of hard shutdowns.
+- You can now monitor finished jobs by checking `FinishedJobRegistry`.
+  Thanks Nic Cope for helping!
+- Jobs with unmet dependencies are now created with `deferred` as their
+  status. You can monitor deferred jobs by checking `DeferredJobRegistry`.
+- It is now possible to enqueue a job at the beginning of queue using
+  `queue.enqueue(func, at_front=True)`. Thanks Travis Johnson!
+- Command line scripts have all been refactored to use `click`. Thanks Lyon Zhang!
+- Added a new `SimpleWorker` that does not fork when executing jobs.
+  Useful for testing purposes. Thanks Cal Leeming!
+- Added `--queue-class` and `--job-class` arguments to `rqworker` script.
+  Thanks David Bonner!
+- Many other minor bug fixes and enhancements.
+
+
 ### 0.4.6
 (May 21st, 2014)
 

From e8c3b96a5a9f4f28a3fc8a42eb330b91af627b7f Mon Sep 17 00:00:00 2001
From: Vincent Driessen <vincent@3rdcloud.com>
Date: Fri, 30 Jan 2015 09:17:34 +0100
Subject: [PATCH 10/16] Bump to 0.5.0.

---
 rq/version.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/rq/version.py b/rq/version.py
index 28d738f..30d837a 100644
--- a/rq/version.py
+++ b/rq/version.py
@@ -1,4 +1,4 @@
 # -*- coding: utf-8 -*-
 from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
-VERSION = '0.4.6'
+VERSION = '0.5.0'

From fac2b10309a7148b13cd259e6770f2695c126628 Mon Sep 17 00:00:00 2001
From: Vincent Driessen <vincent@3rdcloud.com>
Date: Fri, 30 Jan 2015 09:22:53 +0100
Subject: [PATCH 11/16] Auto-sort imports using isort.

---
 rq/cli/helpers.py    |  3 +--
 rq/job.py            |  2 +-
 rq/queue.py          | 11 +++++------
 rq/utils.py          |  4 ++--
 rq/worker.py         |  6 +++---
 tests/test_job.py    | 12 ++++++------
 tests/test_queue.py  |  8 ++++----
 tests/test_worker.py | 14 +++++++-------
 8 files changed, 29 insertions(+), 31 deletions(-)

diff --git a/rq/cli/helpers.py b/rq/cli/helpers.py
index 9ea4b58..0730d17 100644
--- a/rq/cli/helpers.py
+++ b/rq/cli/helpers.py
@@ -8,9 +8,8 @@ from functools import partial
 
 import click
 from rq import Queue, Worker
-from rq.worker import WorkerStatus
 from rq.logutils import setup_loghandlers
-from rq.suspension import is_suspended
+from rq.worker import WorkerStatus
 
 red = partial(click.style, fg='red')
 green = partial(click.style, fg='green')
diff --git a/rq/job.py b/rq/job.py
index d815c5b..981694c 100644
--- a/rq/job.py
+++ b/rq/job.py
@@ -12,7 +12,7 @@ from rq.compat import as_text, decode_redis_hash, string_types, text_type
 from .connections import resolve_connection
 from .exceptions import NoSuchJobError, UnpickleError
 from .local import LocalStack
-from .utils import import_attribute, utcformat, utcnow, utcparse, enum
+from .utils import enum, import_attribute, utcformat, utcnow, utcparse
 
 try:
     import cPickle as pickle
diff --git a/rq/queue.py b/rq/queue.py
index e2c358c..9815f7e 100644
--- a/rq/queue.py
+++ b/rq/queue.py
@@ -4,15 +4,14 @@ from __future__ import (absolute_import, division, print_function,
 
 import uuid
 
-from .connections import resolve_connection
-from .job import Job, JobStatus
-from .utils import import_attribute, utcnow
+from redis import WatchError
 
+from .compat import as_text, string_types, total_ordering
+from .connections import resolve_connection
 from .exceptions import (DequeueTimeout, InvalidJobOperationError,
                          NoSuchJobError, UnpickleError)
-from .compat import total_ordering, string_types, as_text
-
-from redis import WatchError
+from .job import Job, JobStatus
+from .utils import import_attribute, utcnow
 
 
 def get_failed_queue(connection=None):
diff --git a/rq/utils.py b/rq/utils.py
index db56020..7bcb392 100644
--- a/rq/utils.py
+++ b/rq/utils.py
@@ -9,12 +9,12 @@ from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
 
 import calendar
-import importlib
 import datetime
+import importlib
 import logging
 import sys
 
-from .compat import is_python_version, as_text
+from .compat import as_text, is_python_version
 
 
 class _Colorizer(object):
diff --git a/rq/worker.py b/rq/worker.py
index b971a24..0c53409 100644
--- a/rq/worker.py
+++ b/rq/worker.py
@@ -21,11 +21,11 @@ from .exceptions import DequeueTimeout, NoQueueError
 from .job import Job, JobStatus
 from .logutils import setup_loghandlers
 from .queue import get_failed_queue, Queue
-from .timeouts import UnixSignalDeathPenalty
-from .utils import import_attribute, make_colorizer, utcformat, utcnow, enum
-from .version import VERSION
 from .registry import FinishedJobRegistry, StartedJobRegistry
 from .suspension import is_suspended
+from .timeouts import UnixSignalDeathPenalty
+from .utils import enum, import_attribute, make_colorizer, utcformat, utcnow
+from .version import VERSION
 
 try:
     from procname import setprocname
diff --git a/tests/test_job.py b/tests/test_job.py
index b7854b3..89adf6d 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -4,18 +4,18 @@ from __future__ import (absolute_import, division, print_function,
 
 from datetime import datetime
 
+from tests import RQTestCase
+from tests.fixtures import (access_self, CallableObject, Number, say_hello,
+                            some_calculation)
+from tests.helpers import strip_microseconds
+
 from rq.compat import as_text, PY2
 from rq.exceptions import NoSuchJobError, UnpickleError
 from rq.job import get_current_job, Job
-from rq.registry import DeferredJobRegistry
 from rq.queue import Queue
+from rq.registry import DeferredJobRegistry
 from rq.utils import utcformat
 
-from tests import RQTestCase
-from tests.fixtures import (access_self, CallableObject, Number, say_hello,
-                            some_calculation)
-from tests.helpers import strip_microseconds
-
 try:
     from cPickle import loads, dumps
 except ImportError:
diff --git a/tests/test_queue.py b/tests/test_queue.py
index 369b3f1..175a1c0 100644
--- a/tests/test_queue.py
+++ b/tests/test_queue.py
@@ -2,16 +2,16 @@
 from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
 
+from tests import RQTestCase
+from tests.fixtures import (div_by_zero, echo, Number, say_hello,
+                            some_calculation)
+
 from rq import get_failed_queue, Queue
 from rq.exceptions import InvalidJobOperationError
 from rq.job import Job, JobStatus
 from rq.registry import DeferredJobRegistry
 from rq.worker import Worker
 
-from tests import RQTestCase
-from tests.fixtures import (div_by_zero, echo, Number, say_hello,
-                            some_calculation)
-
 
 class CustomJob(Job):
     pass
diff --git a/tests/test_worker.py b/tests/test_worker.py
index d3b8f41..8a35d03 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -5,17 +5,17 @@ from __future__ import (absolute_import, division, print_function,
 import os
 from time import sleep
 
-from rq import get_failed_queue, Queue, Worker, SimpleWorker
-from rq.compat import as_text
-from rq.job import Job, JobStatus
-from rq.registry import StartedJobRegistry
-from rq.suspension import suspend, resume
-
 from tests import RQTestCase, slow
 from tests.fixtures import (create_file, create_file_after_timeout,
-                            div_by_zero, say_hello, say_pid, do_nothing)
+                            div_by_zero, do_nothing, say_hello, say_pid)
 from tests.helpers import strip_microseconds
 
+from rq import get_failed_queue, Queue, SimpleWorker, Worker
+from rq.compat import as_text
+from rq.job import Job, JobStatus
+from rq.registry import StartedJobRegistry
+from rq.suspension import resume, suspend
+
 
 class CustomJob(Job):
     pass

From b8d425b318567073a8343b8f884547b62ced810d Mon Sep 17 00:00:00 2001
From: Vincent Driessen <vincent@3rdcloud.com>
Date: Fri, 30 Jan 2015 09:23:07 +0100
Subject: [PATCH 12/16] Various PEP8 fixes.

---
 rq/queue.py          | 1 -
 rq/suspension.py     | 2 +-
 rq/utils.py          | 2 +-
 rq/version.py        | 1 +
 rq/worker.py         | 7 ++-----
 tests/test_cli.py    | 6 ------
 tests/test_job.py    | 2 +-
 tests/test_queue.py  | 2 +-
 tests/test_worker.py | 2 +-
 9 files changed, 8 insertions(+), 17 deletions(-)

diff --git a/rq/queue.py b/rq/queue.py
index 9815f7e..3f59a50 100644
--- a/rq/queue.py
+++ b/rq/queue.py
@@ -253,7 +253,6 @@ class Queue(object):
 
         If Queue is instantiated with async=False, job is executed immediately.
         """
-        
         with self.connection._pipeline() as pipeline:
             # Add Queue key set
             self.connection.sadd(self.redis_queues_keys, self.key)
diff --git a/rq/suspension.py b/rq/suspension.py
index b734acd..93152b9 100644
--- a/rq/suspension.py
+++ b/rq/suspension.py
@@ -15,4 +15,4 @@ def suspend(connection, ttl=None):
 
 
 def resume(connection):
-    return connection.delete(WORKERS_SUSPENDED)
\ No newline at end of file
+    return connection.delete(WORKERS_SUSPENDED)
diff --git a/rq/utils.py b/rq/utils.py
index 7bcb392..3e44a98 100644
--- a/rq/utils.py
+++ b/rq/utils.py
@@ -217,4 +217,4 @@ def enum(name, *sequential, **named):
     # On Python 2 type() requires a byte string (which is str() on Python 2).
     # On Python 3 it does not matter, so we'll use str(), which acts as
     # a no-op.
-    return type(str(name), (), values)
\ No newline at end of file
+    return type(str(name), (), values)
diff --git a/rq/version.py b/rq/version.py
index 30d837a..0d96ce8 100644
--- a/rq/version.py
+++ b/rq/version.py
@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
 from __future__ import (absolute_import, division, print_function,
                         unicode_literals)
+
 VERSION = '0.5.0'
diff --git a/rq/worker.py b/rq/worker.py
index 0c53409..201dfed 100644
--- a/rq/worker.py
+++ b/rq/worker.py
@@ -12,7 +12,6 @@ import sys
 import time
 import traceback
 import warnings
-from datetime import datetime
 
 from rq.compat import as_text, string_types, text_type
 
@@ -54,8 +53,8 @@ def compact(l):
     return [x for x in l if x is not None]
 
 _signames = dict((getattr(signal, signame), signame)
-    for signame in dir(signal)
-    if signame.startswith('SIG') and '_' not in signame)
+                 for signame in dir(signal)
+                 if signame.startswith('SIG') and '_' not in signame)
 
 
 def signal_name(signum):
@@ -367,7 +366,6 @@ class Worker(object):
         if before_state:
             self.set_state(before_state)
 
-
     def work(self, burst=False):
         """Starts the work loop.
 
@@ -416,7 +414,6 @@ class Worker(object):
                 self.register_death()
         return did_perform_work
 
-
     def dequeue_job_and_maintain_ttl(self, timeout):
         result = None
         qnames = self.queue_names()
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 3977006..f1b7fd4 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -101,9 +101,3 @@ class TestRQCli(RQTestCase):
 
         self.assertEqual(result.exit_code, 1)
         self.assertIn("Duration must be an integer greater than 1", result.output)
-
-
-
-
-
-
diff --git a/tests/test_job.py b/tests/test_job.py
index 89adf6d..e117ea9 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -339,7 +339,7 @@ class TestJob(RQTestCase):
         job = Job.create(func=say_hello, origin=origin)
         job._dependency_id = 'id'
         job.save()
-        
+
         self.assertEqual(registry.get_job_ids(), [])
         job.register_dependency()
         self.assertEqual(as_text(self.testconn.spop('rq:job:id:dependents')), job.id)
diff --git a/tests/test_queue.py b/tests/test_queue.py
index 175a1c0..e4e9253 100644
--- a/tests/test_queue.py
+++ b/tests/test_queue.py
@@ -340,7 +340,7 @@ class TestQueue(RQTestCase):
         self.assertFalse(self.testconn.exists(parent_job.dependents_key))
 
         # DeferredJobRegistry should also be empty
-        self.assertEqual(registry.get_job_ids(), [])        
+        self.assertEqual(registry.get_job_ids(), [])
 
     def test_enqueue_job_with_dependency(self):
         """Jobs are enqueued only when their dependencies are finished."""
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 8a35d03..c47a62e 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -334,7 +334,7 @@ class TestWorker(RQTestCase):
                 raise
 
         q = Queue()
-        job = q.enqueue(create_file, SENTINEL_FILE)
+        q.enqueue(create_file, SENTINEL_FILE)
 
         w = Worker([q])
 

From 1018330cea8ac5e6fe407860f012c6c90e279e62 Mon Sep 17 00:00:00 2001
From: Leonid Shvechikov <shvechikov@gmail.com>
Date: Sun, 1 Feb 2015 10:37:21 +0300
Subject: [PATCH 13/16] Tiny fix of the indefinite article before "RQ"

"An" should be used here because "RQ" sounds like "ar-kiu".
---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 8b83f68..2b1fa71 100644
--- a/README.md
+++ b/README.md
@@ -33,7 +33,7 @@ def count_words_at_url(url):
 
 You do use the excellent [requests][r] package, don't you?
 
-Then, create a RQ queue:
+Then, create an RQ queue:
 
 ```python
 from rq import Queue, use_connection

From c155918d6de1a781a3805e6f1e1b6f866c9a3a88 Mon Sep 17 00:00:00 2001
From: glaslos <lukas.rist@bluecoat.com>
Date: Wed, 18 Feb 2015 15:36:39 +0100
Subject: [PATCH 14/16] lrem only first occurrence of the job  id (we expect
 only one)

---
 rq/queue.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/rq/queue.py b/rq/queue.py
index 3f59a50..988a54e 100644
--- a/rq/queue.py
+++ b/rq/queue.py
@@ -142,9 +142,9 @@ class Queue(object):
         job_id = job_or_id.id if isinstance(job_or_id, self.job_class) else job_or_id
 
         if pipeline is not None:
-            pipeline.lrem(self.key, 0, job_id)
+            pipeline.lrem(self.key, 1, job_id)
 
-        return self.connection._lrem(self.key, 0, job_id)
+        return self.connection._lrem(self.key, 1, job_id)
 
     def compact(self):
         """Removes all "dead" jobs from the queue by cycling through it, while

From 100522237187d3aef45a54b9bdf4e3ff095fa173 Mon Sep 17 00:00:00 2001
From: Selwin Ong <selwin.ong@gmail.com>
Date: Mon, 23 Feb 2015 17:28:51 +0700
Subject: [PATCH 15/16] Restored the ability to specify connection params in
 config

---
 rq/cli/cli.py         | 20 ++++++++++++--------
 rq/cli/helpers.py     | 15 +++++++++++++++
 tests/test_helpers.py | 41 +++++++++++++++++++++++++++++++++++++++++
 3 files changed, 68 insertions(+), 8 deletions(-)
 create mode 100644 tests/test_helpers.py

diff --git a/rq/cli/cli.py b/rq/cli/cli.py
index 8dd0b2b..d0a5eff 100755
--- a/rq/cli/cli.py
+++ b/rq/cli/cli.py
@@ -16,22 +16,27 @@ from rq import Connection, get_failed_queue, Queue
 from rq.contrib.legacy import cleanup_ghosts
 from rq.exceptions import InvalidJobOperationError
 from rq.utils import import_attribute
-from rq.suspension import suspend as connection_suspend, resume as connection_resume, is_suspended
+from rq.suspension import (suspend as connection_suspend,
+                           resume as connection_resume, is_suspended)
 
-from .helpers import (read_config_file, refresh, setup_loghandlers_from_args,
-                      show_both, show_queues, show_workers)
+from .helpers import (get_redis_from_config, read_config_file, refresh,
+                      setup_loghandlers_from_args, show_both, show_queues,
+                      show_workers)
 
 
 url_option = click.option('--url', '-u', envvar='RQ_REDIS_URL',
                           help='URL describing Redis connection details.')
 
-config_option = click.option('--config', '-c', help='Module containing RQ settings.')
+config_option = click.option('--config', '-c',
+                             help='Module containing RQ settings.')
 
 
 def connect(url, config=None):
+    if url:
+        return StrictRedis.from_url(url)
+
     settings = read_config_file(config) if config else {}
-    url = url or settings.get('REDIS_URL')
-    return StrictRedis.from_url(url or 'redis://localhost:6379/0')
+    return get_redis_from_config(settings)
 
 
 @click.group()
@@ -148,7 +153,6 @@ def worker(url, config, burst, name, worker_class, job_class, queue_class, path,
 
     settings = read_config_file(config) if config else {}
     # Worker specific default arguments
-    url = url or settings.get('REDIS_URL')
     queues = queues or settings.get('QUEUES', ['default'])
     sentry_dsn = sentry_dsn or settings.get('SENTRY_DSN')
 
@@ -158,7 +162,7 @@ def worker(url, config, burst, name, worker_class, job_class, queue_class, path,
 
     setup_loghandlers_from_args(verbose, quiet)
 
-    conn = connect(url)
+    conn = connect(url, config)
     cleanup_ghosts(conn)
     worker_class = import_attribute(worker_class)
     queue_class = import_attribute(queue_class)
diff --git a/rq/cli/helpers.py b/rq/cli/helpers.py
index 0730d17..d5c54a1 100644
--- a/rq/cli/helpers.py
+++ b/rq/cli/helpers.py
@@ -7,6 +7,8 @@ import time
 from functools import partial
 
 import click
+from redis import StrictRedis
+
 from rq import Queue, Worker
 from rq.logutils import setup_loghandlers
 from rq.worker import WorkerStatus
@@ -24,6 +26,19 @@ def read_config_file(module):
                  if k.upper() == k])
 
 
+def get_redis_from_config(settings):
+    """Returns a StrictRedis instance from a dictionary of settings."""
+    if settings.get('REDIS_URL') is not None:
+        return StrictRedis.from_url(settings['REDIS_URL'])
+
+    return StrictRedis(
+        host=settings.get('REDIS_HOST', 'localhost'),
+        port=settings.get('REDIS_PORT', 6379),
+        db=settings.get('REDIS_DB', 0),
+        password=settings.get('REDIS_PASSWORD', None),
+    )
+
+
 def pad(s, pad_to_length):
     """Pads the given string to the given length."""
     return ('%-' + '%ds' % pad_to_length) % (s,)
diff --git a/tests/test_helpers.py b/tests/test_helpers.py
new file mode 100644
index 0000000..b43f13b
--- /dev/null
+++ b/tests/test_helpers.py
@@ -0,0 +1,41 @@
+from rq.cli.helpers import get_redis_from_config
+
+from tests import RQTestCase
+
+
+class TestHelpers(RQTestCase):
+
+    def test_get_redis_from_config(self):
+        """Ensure Redis connection params are properly parsed"""
+        settings = {
+            'REDIS_URL': 'redis://localhost:1/1'
+        }
+
+        # Ensure REDIS_URL is read
+        redis = get_redis_from_config(settings)
+        connection_kwargs = redis.connection_pool.connection_kwargs
+        self.assertEqual(connection_kwargs['db'], 1)
+        self.assertEqual(connection_kwargs['port'], 1)
+
+        settings = {
+            'REDIS_URL': 'redis://localhost:1/1',
+            'REDIS_HOST': 'foo',
+            'REDIS_DB': 2,
+            'REDIS_PORT': 2,
+            'REDIS_PASSWORD': 'bar'
+        }
+
+        # Ensure REDIS_URL is preferred
+        redis = get_redis_from_config(settings)
+        connection_kwargs = redis.connection_pool.connection_kwargs
+        self.assertEqual(connection_kwargs['db'], 1)
+        self.assertEqual(connection_kwargs['port'], 1)
+
+        # Ensure fall back to regular connection parameters
+        settings['REDIS_URL'] = None
+        redis = get_redis_from_config(settings)
+        connection_kwargs = redis.connection_pool.connection_kwargs
+        self.assertEqual(connection_kwargs['host'], 'foo')
+        self.assertEqual(connection_kwargs['db'], 2)
+        self.assertEqual(connection_kwargs['port'], 2)
+        self.assertEqual(connection_kwargs['password'], 'bar')

From 219f21b6378bbd66a13e997876c7c7416b27ef46 Mon Sep 17 00:00:00 2001
From: alternativshik <alternativshik@gmail.com>
Date: Thu, 25 Sep 2014 10:47:17 +0700
Subject: [PATCH 16/16] Allow non-ASCII characters in arguments

get_call_string() failed if any arguments contained non-ASCII strings.

Fixes #406
---
 rq/job.py         | 11 +++++++++--
 tests/test_job.py | 20 ++++++++++++++++++++
 2 files changed, 29 insertions(+), 2 deletions(-)

diff --git a/rq/job.py b/rq/job.py
index 981694c..5dbbcf5 100644
--- a/rq/job.py
+++ b/rq/job.py
@@ -514,8 +514,15 @@ class Job(object):
         if self.func_name is None:
             return None
 
-        arg_list = [repr(arg) for arg in self.args]
-        arg_list += ['%s=%r' % (k, v) for k, v in self.kwargs.items()]
+        # Python 2/3 compatibility
+        try:
+            arg_list = [repr(arg).decode('utf-8') for arg in self.args]
+        except AttributeError:
+            arg_list = [repr(arg) for arg in self.args]
+
+        kwargs = ['{0}={1!r}'.format(k, v) for k, v in self.kwargs.items()]
+        # Sort here because python 3.3 & 3.4 makes different call_string
+        arg_list += sorted(kwargs)
         args = ', '.join(arg_list)
         return '%s(%s)' % (self.func_name, args)
 
diff --git a/tests/test_job.py b/tests/test_job.py
index e117ea9..d2e7f44 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -23,6 +23,26 @@ except ImportError:
 
 
 class TestJob(RQTestCase):
+    def test_unicode(self):
+        """Unicode in job description [issue405]"""
+        job = Job.create(
+            'myfunc',
+            args=[12, "☃"],
+            kwargs=dict(snowman="☃", null=None),
+        )
+
+        try:
+            # Python 2
+            test_string = u"myfunc(12, u'\\u2603', null=None, snowman=u'\\u2603')".decode('utf-8')
+        except AttributeError:
+            # Python 3
+            test_string = "myfunc(12, '☃', null=None, snowman='☃')"
+
+        self.assertEquals(
+            job.description,
+            test_string,
+        )
+
     def test_create_empty_job(self):
         """Creation of new empty jobs."""
         job = Job()