Rhumba is an asynchronous job queue and distributed task service with scheduling.
$ pip install rhumbaRhumba tasks are created as plugins configured on each Rhumba worker. The
plugin must contain a class called Plugin and callable task methods should be
prefixed with call_
from rhumba import RhumbaPlugin
class Plugin(RhumbaPlugin):
def call_test(self, args):
self.log("Test call %s" % repr(args))A plugin is connected to a queue in the Rhumba YAML configuration file
queues:
- name: myqueue
plugin: mypluginPlugins must exist in the PYTHON_PATH of the process
Then start up your Rhumba worker with twistd -n rhumba -c myconfig.yaml
(ve)colin@nemesis:~/rhumba$ twistd -n rhumba -c examples/rhumba.yml
2015-11-08 23:15:20+0200 [-] Log opened.
2015-11-08 23:15:20+0200 [-] twistd 15.4.0 (/home/colin/rhumba/ve/bin/python 2.7.6) starting up.
2015-11-08 23:15:20+0200 [-] reactor class: twisted.internet.epollreactor.EPollReactor.
2015-11-08 23:15:20+0200 [-] Starting Rhumba
2015-11-08 23:15:20+0200 [-] Starting queue testqueue: plugin=<examples.testplugin.Plugin object at 0x7f0836655190>
2015-11-08 23:15:21+0200 [RedisClient,client] Queing testqueue scheduled job {'message': 'crontest', 'version': 1, 'params': {}, 'id': 'd1f23d42865d11e58d6482576555349e'}
2015-11-08 23:15:21+0200 [RedisClient,client] [testqueue]: tick!Task methods can return a deferred, do not implement blocking tasks.
from twisted.mail.smtp import sendmail
from rhumba import RhumbaPlugin
class Plugin(RhumbaPlugin):
def mailSent(self, from, to, msg):
self.log('Mail from %s to %s sent successfully' % (from, to))
def call_sendemail(self, args):
a = (args['from'], args['to'], args['message'])
d = sendmail('localhost', *a)
d.addCallback(self.mailSent, *a)
return dRhumbas @cron decorator can be used to schedule tasks.
from rhumba import RhumbaPlugin, cron
class Plugin(RhumbaPlugin):
@cron(hour=12)
def call_scheduled(self, args):
self.log("It's 12pm!")The cron decorator accepts 5 keyword arguments: secs, min, hour, day,
month and weekday. All of these arguments except weekday can also accept
a string of the format "*/n" in which case the task will be run every n
seconds, minutes, hours, days or months. The arguments may contain any
combination of requirements as long as they make sense.
For example you may have a task run every 5 minutes from 1pm to 2pm on Sundays in June.
from rhumba import RhumbaPlugin, cron, crontab
class Plugin(RhumbaPlugin):
@cron(mins="*/5", hour=13, month=6, weekday=crontab.Sunday)
def call_scheduled(self, args):
self.log("It's 12pm!")Matching ranges and lists is also possible, eg hour="1-3" or day=(5, 6, 9)
and any combination of these.
from rhumba import RhumbaPlugin, cron, crontab
class Plugin(RhumbaPlugin):
@cron(hour="8-17", weekday=crontab.Weekdays)
def call_during_business_hours(self, args):
# do stuff Tasks can also return JSON serialisable results.
Rhumba provides a synchronous client suitable for use with frameworks like Django as an alternative to Celery, as well as an async client for Twisted.
from django.shortcuts import redirect
# etc...
from rhumba.client import RhumbaClient
def sendemail(request):
if request.method == "POST":
form = SomeEmailForm(request.POST)
if form.is_valid():
c = RhumbaClient()
c.queue('myqueue', 'sendemail', form.cleaned_data)
return redirect('index')You can not send Django ORM objects down the wire to Rhumba, if you need to access your database you have to reimplement those queries in Twisted
You can retrieve the result of a task (if one is returned) until it expires
using the getResult method. Expiry can be set in the workers queue
configuration using the expire option and defaults to 1 hour.
def sendemail(request):
if request.method == "POST":
form = SomeEmailForm(request.POST)
if form.is_valid():
c = RhumbaClient()
taskid = c.queue('myqueue', 'sendemail', form.cleaned_data)
return redirect('checkmail', id=taskid)
def checkmail(request, id):
# This is a really silly example, do something sane instead...
r = RhumbaClient().getResult('myqueue', id)
if r:
return redirect('index')
else:
return redirect('checkmail', id=id)The async client rhumba.client.AsyncRhumbaClient provides the same API but returns deferreds as expected using the txredis client.
As with all new fangled software Rhumba provides an HTTP API on port 7701 which will return stats about queues and workers in the pool and also provides a convenient way to queue tasks and wait for them without connecting directly to the backend.
{
"workers": {
"cthulhu": [
{
"status": "ready",
"lastseen": 1448567076.76,
"id": "ffc0dc9e947511e59f38448a5b5fd8c0"
}
]
},
"queues": {
"testqueue": {
"waiting": 0,
"messages": {
"crontest": {
"count": 345,
"time": 22.99
},
"test": {
"count": 6,
"time": 52000.62
}
}
}
},
"crons": {
"testqueue": {
"master": "ffc0dc9e947511e59f38448a5b5fd8c0:cthulhu",
"methods": {
"call_crontest": 1448567073.0
}
}
}
}["testqueue"]{
"uid": "15b81fb6947711e58154448a5b5fd8c0"
}{
"result": ["Hello!"],
"time": 1448567500.198432
}{
"result": ["Hello!"],
"time": 1448567475.792436
}
Both wait and call accept HTTP POST requests with a json payload
{
"result": ["Hello, World!"],
"time": 1448567851.689572
}{
"uid": "15b81fb6947711e58154448a5b5fd8c0"
}Makes a fanout request to all servers active on a specific queue
{
"511c043cbcf411e5abaa82576555349e": {
"result": ["Hello, test!"],
"time": 1453019709.481646
},
"f363fe66bcf411e597fa448a5b5fd8c0": {
"result": ["Hello, test!"],
"time": 1453019597.918752
}
}Makes a fanout request to all servers and waits for all of them to return a result