def consume_items(queue, callback, verbose=True):
 from pylons import tmpl_context as c
 chan = connection_manager.get_channel()
 chan.basic_qos(prefetch_size=0, prefetch_count=10, a_global=False)
 def _callback(msg):
  if verbose:
   count_str = ''
   if ('message_count' in msg.delivery_info):
    count_str = ('(%d remaining)' % msg.delivery_info['message_count'])
   print ('%s: 1 item %s' % (queue, count_str))
  cfg.reset_caches()
  c.use_write_db = {}
  ret = callback(msg)
  msg.channel.basic_ack(msg.delivery_tag)
  sys.stdout.flush()
  return ret
 chan.basic_consume(queue=queue, callback=_callback)
 try:
  while chan.callbacks:
   try:
    chan.wait()
   except KeyboardInterrupt:
    break
 finally:
  worker.join()
  if chan.is_open:
   chan.close()

def consume_items ( queue , callback , verbose = True ) : from pylons import tmpl_context as c chan = connection_manager . get_channel ( ) chan . basic_qos ( prefetch_size = 0 , prefetch_count = 10 , a_global = False ) def _callback ( msg ) : if verbose : count_str = '' if ( 'message_count' in msg . delivery_info ) : count_str = ( '(%d remaining)' % msg . delivery_info [ 'message_count' ] ) print ( '%s: 1 item %s' % ( queue , count_str ) ) cfg . reset_caches ( ) c . use_write_db = { } ret = callback ( msg ) msg . channel . basic_ack ( msg . delivery_tag ) sys . stdout . flush ( ) return ret chan . basic_consume ( queue = queue , callback = _callback ) try : while chan . callbacks : try : chan . wait ( ) except KeyboardInterrupt : break finally : worker . join ( ) if chan . is_open : chan . close ( )
a lighter-weight version of handle_items that uses amqps basic .