-
Notifications
You must be signed in to change notification settings - Fork 2
/
redisutil.py
54 lines (47 loc) · 1.47 KB
/
redisutil.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import collections
import logging
import time
import tornado
import tornadoredis
@tornado.gen.engine
def new_redis(callback=None):
while True:
try:
logging.info('Establishing redis connection')
redis = tornadoredis.Client()
redis.connect()
callback(redis)
break
except:
loop = tornado.ioloop.IOLoop.instance()
yield tornado.gen.Task(loop.add_timeout, time.time() + 1)
class RedisMQ(object):
def __init__(self, name):
self.name = name
self.queue = collections.deque()
@tornado.gen.engine
def connect(self, callback=None):
self.redis = None
self.redis = yield tornado.gen.Task(new_redis)
while self.queue:
message = self.queue.popleft()
self.send(message)
callback()
def send(self, message):
try:
self.redis.rpush(self.name, message)
print '->', repr(message)
except:
self.queue.append(message)
if self.redis is not None:
self.connect()
@tornado.gen.engine
def loop(self, handler):
while True:
response = yield tornado.gen.Task(self.redis.blpop, self.name)
if isinstance(response, Exception):
yield tornado.gen.Task(self.connect)
else:
message = response[self.name]
print '<-', repr(message)
handler(message)