def _add_item(routing_key, body, message_id=None, delivery_mode=DELIVERY_DURABLE, headers=None, exchange=None, send_stats=True):
 if (not cfg.amqp_host):
  cfg.log.error(('Ignoring amqp message %r to %r' % (body, routing_key)))
  return
 if (not exchange):
  exchange = cfg.amqp_exchange
 chan = connection_manager.get_channel()
 msg = amqp.Message(body, timestamp=datetime.now(), delivery_mode=delivery_mode)
 if message_id:
  msg.properties['message_id'] = message_id
 if headers:
  msg.properties['application_headers'] = headers
 event_name = ('amqp.%s' % routing_key)
 try:
  chan.basic_publish(msg, exchange=exchange, routing_key=routing_key)
 except Exception as e:
  if send_stats:
   cfg.stats.event_count(event_name, 'enqueue_failed')
  if (e.errno == errno.EPIPE):
   connection_manager.get_channel(True)
   add_item(routing_key, body, message_id)
  else:
   raise
 else:
  if send_stats:
   cfg.stats.event_count(event_name, 'enqueue')

def _add_item ( routing_key , body , message_id = None , delivery_mode = DELIVERY_DURABLE , headers = None , exchange = None , send_stats = True ) : if ( not cfg . amqp_host ) : cfg . log . error ( ( 'Ignoring amqp message %r to %r' % ( body , routing_key ) ) ) return if ( not exchange ) : exchange = cfg . amqp_exchange chan = connection_manager . get_channel ( ) msg = amqp . Message ( body , timestamp = datetime . now ( ) , delivery_mode = delivery_mode ) if message_id : msg . properties [ 'message_id' ] = message_id if headers : msg . properties [ 'application_headers' ] = headers event_name = ( 'amqp.%s' % routing_key ) try : chan . basic_publish ( msg , exchange = exchange , routing_key = routing_key ) except Exception as e : if send_stats : cfg . stats . event_count ( event_name , 'enqueue_failed' ) if ( e . errno == errno . EPIPE ) : connection_manager . get_channel ( True ) add_item ( routing_key , body , message_id ) else : raise else : if send_stats : cfg . stats . event_count ( event_name , 'enqueue' )
adds an item onto a queue .