I am looking for a way to use kombu as MQ adapter between tornado-sockjs and Django application server. I did something like:
class BrokerClient(ConsumerMixin):
clients = []
def __init__(self):
self.connection = BrokerConnection(settings.BROKER_URL)
self.io_loop = ioloop.IOLoop.instance()
self.queue = sockjs_queue
self._handle_loop()
@staticmethod
def instance():
if not hasattr(BrokerClient, '_instance'):
BrokerClient._instance = BrokerClient()
return BrokerClient._instance
def add_client(self, client):
self.clients.append(client)
def remove_client(self, client):
self.clients.remove(client)
def _handle_loop(self):
try:
if self.restart_limit.can_consume(1):
for _ in self.consume(limit=5):
pass
except self.connection.connection_errors:
print ('Connection to broker lost. '
'Trying to re-establish the connection...')
self.io_loop.add_timeout(datetime.timedelta(0.0001), self._handle_loop)
def get_consumers(self, Consumer, channel):
return [Consumer([self.queue, ], callbacks=[self.process_task])]
def process_task(self, body, message):
for client in self.clients:
if hasattr(body, 'users') and client.user.pk in body.users:
client.send(body)
message.ack()
But tornado blocked at _handle_loop execution (as expected).
Is there any way to prevent this?
I am aware of Pika library adapter for Tornado, but I would like to use kombu because it's already used in project and has flexible transports.
UPDATE:
Changed _handle_loop to generator function
def drain_events(self, callback):
with self.Consumer() as (connection, channel, consumers):
with self.extra_context(connection, channel):
try:
connection.drain_events(timeout=1)
except:
pass
callback(None)
@tornado.gen.engine
def _handle_loop(self):
response = yield tornado.gen.Task(self.drain_events)
self.io_loop.add_timeout(datetime.timedelta(0.0001), self._handle_loop)
Finally I found the correct solution for RabbitMQ backend:
class BrokerClient(object):
clients = []
@staticmethod
def instance():
if not hasattr(BrokerClient, '_instance'):
BrokerClient._instance = BrokerClient()
return BrokerClient._instance
def __init__(self):
self.connection = BrokerConnection(settings.BROKER_URL)
self.consumer = Consumer(self.connection.channel(), [queue, ], callbacks=[self.process_task])
self.consumer.consume()
io_loop = tornado.ioloop.IOLoop.instance()
for sock, handler in self.connection.eventmap.items():
def drain_nowait(fd, events):
handler()
io_loop.add_handler(sock.fileno(), drain_nowait, l.READ | l.ERROR)
def process_task(self, body, message):
#something
message.ack()
def add_client(self, client):
self.clients.append(client)
def remove_client(self, client):
self.clients.remove(client)
For other backends, you can use the solution posted in the question
NOTE: Doesn't work with librabbitmq
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With