mirror of https://github.com/rq/rq.git
Add ability to put work on alternate queues.
This commit is contained in:
parent
1b8da4a861
commit
407f3e8b38
12
rq/job.py
12
rq/job.py
|
@ -24,13 +24,19 @@ class job(object):
|
|||
self.queue = Queue(queue_name)
|
||||
|
||||
def __call__(self, f):
|
||||
def delay(*args, **kwargs):
|
||||
rv_key = '%s:result:%s' % (self.queue.key, str(uuid.uuid4()))
|
||||
def enqueue(queue, *args, **kwargs):
|
||||
if not isinstance(queue, Queue):
|
||||
raise ValueError('Argument queue must be a Queue.')
|
||||
rv_key = '%s:result:%s' % (queue.key, str(uuid.uuid4()))
|
||||
if f.__module__ == '__main__':
|
||||
raise ValueError('Functions from the __main__ module cannot be processed by workers.')
|
||||
s = dumps((f, rv_key, args, kwargs))
|
||||
conn.rpush(self.queue.key, s)
|
||||
conn.rpush(queue.key, s)
|
||||
return DelayedResult(rv_key)
|
||||
f.enqueue = enqueue
|
||||
|
||||
def delay(*args, **kwargs):
|
||||
return f.enqueue(self.queue, *args, **kwargs)
|
||||
f.delay = delay
|
||||
return f
|
||||
|
||||
|
|
|
@ -17,11 +17,18 @@ class Queue(object):
|
|||
|
||||
@property
|
||||
def empty(self):
|
||||
return conn.llen(self.key) == 0
|
||||
return self.count == 0
|
||||
|
||||
@property
|
||||
def messages(self):
|
||||
return conn.lrange(self.key, 0, -1)
|
||||
|
||||
@property
|
||||
def count(self):
|
||||
return conn.llen(self.key)
|
||||
|
||||
def enqueue(self, job):
|
||||
return job.enqueue(self)
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
|
|
@ -5,8 +5,10 @@ from redis import Redis
|
|||
from rq import conn, Queue, job
|
||||
|
||||
@job('my-queue')
|
||||
def testjob():
|
||||
return 'hi there'
|
||||
def testjob(name=None):
|
||||
if name is None:
|
||||
name = 'Stranger'
|
||||
return 'Hi there, %s!' % (name,)
|
||||
|
||||
|
||||
class RQTestCase(unittest.TestCase):
|
||||
|
@ -62,14 +64,34 @@ class TestQueue(RQTestCase):
|
|||
self.assertEquals(q.empty, False)
|
||||
|
||||
def test_put_work_on_queue(self):
|
||||
"""Putting work on queues."""
|
||||
"""Putting work on queues using delay."""
|
||||
q = Queue('my-queue')
|
||||
self.assertEquals(q.empty, True)
|
||||
|
||||
# testjob spec holds which queue this is sent to
|
||||
testjob.delay()
|
||||
self.assertEquals(q.empty, False)
|
||||
self.assertQueueContains(q, testjob)
|
||||
|
||||
def test_put_work_on_different_queue(self):
|
||||
"""Putting work on alternative queues using enqueue."""
|
||||
|
||||
# Override testjob spec holds which queue
|
||||
q = Queue('different-queue')
|
||||
self.assertEquals(q.empty, True)
|
||||
testjob.enqueue(q)
|
||||
self.assertEquals(q.empty, False)
|
||||
self.assertQueueContains(q, testjob)
|
||||
|
||||
def test_put_work_on_different_queue_reverse(self):
|
||||
"""Putting work on specific queues using the Queue object."""
|
||||
|
||||
q = Queue('alt-queue')
|
||||
self.assertEquals(q.empty, True)
|
||||
q.enqueue(testjob)
|
||||
self.assertEquals(q.empty, False)
|
||||
self.assertQueueContains(q, testjob)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
|
Loading…
Reference in New Issue