def drain_consumer(consumer, limit=1, timeout=None, callbacks=None):
 acc = deque()
 def on_message(body, message):
  acc.append((body, message))
 consumer.callbacks = ([on_message] + (callbacks or []))
 with consumer:
  for _ in eventloop(consumer.channel.connection.client, limit=limit, timeout=timeout, ignore_timeouts=True):
   try:
    (yield acc.popleft())
   except IndexError:
    pass

def drain_consumer ( consumer , limit = 1 , timeout = None , callbacks = None ) : acc = deque ( ) def on_message ( body , message ) : acc . append ( ( body , message ) ) consumer . callbacks = ( [ on_message ] + ( callbacks or [ ] ) ) with consumer : for _ in eventloop ( consumer . channel . connection . client , limit = limit , timeout = timeout , ignore_timeouts = True ) : try : ( yield acc . popleft ( ) ) except IndexError : pass
drain messages from consumer instance .