#!/usr/bin/env python import optparse from rq import use_redis, Queue, Worker from rq import dummy def parse_args(): parser = optparse.OptionParser() parser.add_option('-n', '--count', type='int', dest='count', default=1) opts, args = parser.parse_args() return (opts, args, parser) def main(): import sys sys.path.insert(0, '.') opts, args, parser = parse_args() use_redis() queues = ('default', 'high', 'low') sample_calls = [ (dummy.do_nothing, [], {}), (dummy.sleep, [1], {}), (dummy.fib, [8], {}), # normal result (dummy.fib, [24], {}), # takes pretty long (dummy.div_by_zero, [], {}), # 5 / 0 => div by zero exc (dummy.random_failure, [], {}), # simulate random failure (handy for requeue testing) ] for i in range(opts.count): import random f, args, kwargs = random.choice(sample_calls) q = Queue(random.choice(queues)) q.enqueue(f, *args, **kwargs) #q = Queue('foo') #q.enqueue(do_nothing) #q.enqueue(sleep, 3) #q = Queue('bar') #q.enqueue(yield_stuff) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) #q.enqueue(do_nothing) if __name__ == '__main__': main()