title: Kombu in non-blocking way
body: <p>I am looking for a way to use kombu as MQ adapter between tornado-sockjs and Django application server. I did something like:</p>
  
  <pre><code>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()
  </code></pre>
  
  <p>But tornado blocked at _handle_loop execution (as expected). </p>
  
  <p>Is there any way to prevent this? </p>
  
  <p>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.</p>
  
  <p>UPDATE:</p>
  
  <p>Changed _handle_loop to generator function</p>
  
  <pre><code>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)
  </code></pre>