// NewAMQPJobQueue creates a AMQPJobQueue backed by the given AMQP connections and
// connects to the AMQP queue with the given name. The queue will be declared
// in AMQP when this function is called, so an error could be raised if the
// queue already exists, but with different attributes than we expect.
NewAMQPJobQueue creates a AMQPJobQueue backed by the given AMQP connections and
connects to the AMQP queue with the given name. The queue will be declared
in AMQP when this function is called, so an error could be raised if the
queue already exists, but with different attributes than we expect.
// NewAMQPJobQueue creates a AMQPJobQueue backed by the given AMQP connections and // connects to the AMQP queue with the given name . The queue will be declared // in AMQP when this function is called so an error could be raised if the // queue already exists but with different attributes than we expect .
func NewAMQPJobQueue(conn *amqp.Connection, queue string, stateUpdatePoolSize int, sharded bool) (*AMQPJobQueue, error) {
	channel, err := conn.Channel()
	if err != nil {
		return nil, err
	}
	_, err = channel.QueueDeclare(queue, true, false, false, false, nil)
	if err != nil {
		return nil, err
	}
	err = channel.ExchangeDeclare("reporting", "topic", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}
	_, err = channel.QueueDeclare("reporting.jobs.builds", true, false, false, false, nil)
	if err != nil {
		return nil, err
	}
	if sharded {
		err = channel.ExchangeDeclare("reporting.jobs.logs_sharded", "x-modulus-hash", true, false, false, false, nil)
		if err != nil {
			return nil, err
		}
	} else {
		_, err = channel.QueueDeclare("reporting.jobs.logs", true, false, false, false, nil)
		if err != nil {
			return nil, err
		}
		err = channel.QueueBind("reporting.jobs.logs", "reporting.jobs.logs", "reporting", false, nil)
		if err != nil {
			return nil, err
		}
	}
	err = channel.Close()
	if err != nil {
		return nil, err
	}
	stateUpdatePool := newStateUpdatePool(conn, stateUpdatePoolSize)
	go reportPoolMetrics("state_update_pool", stateUpdatePool)
	return &AMQPJobQueue{
		conn:            conn,
		queue:           queue,
		withLogSharding: sharded,
		stateUpdatePool: stateUpdatePool,
	}, nil
}
func NewAMQPJobQueue ( conn * amqp . Connection , queue string , stateUpdatePoolSize int , sharded bool ) ( * AMQPJobQueue , error ) { channel , err := conn . Channel ( ) if err != nil { return nil , err } _ , err = channel . QueueDeclare ( queue , true , false , false , false , nil ) if err != nil { return nil , err } err = channel . ExchangeDeclare ( " " , " " , true , false , false , false , nil ) if err != nil { return nil , err } _ , err = channel . QueueDeclare ( " " , true , false , false , false , nil ) if err != nil { return nil , err } if sharded { err = channel . ExchangeDeclare ( " " , " " , true , false , false , false , nil ) if err != nil { return nil , err } } else { _ , err = channel . QueueDeclare ( " " , true , false , false , false , nil ) if err != nil { return nil , err } err = channel . QueueBind ( " " , " " , " " , false , nil ) if err != nil { return nil , err } } err = channel . Close ( ) if err != nil { return nil , err } stateUpdatePool := newStateUpdatePool ( conn , stateUpdatePoolSize ) go reportPoolMetrics ( " " , stateUpdatePool ) return & AMQPJobQueue { conn : conn , queue : queue , withLogSharding : sharded , stateUpdatePool : stateUpdatePool , } , nil }
NewAMQPJobQueue creates a AMQPJobQueue backed by the given AMQP connections and
connects to the AMQP queue with the given name.
NewAMQPJobQueue creates a AMQPJobQueue backed by the given AMQP connections and connects to the AMQP queue with the given name .
// This exchange should be declared as sharded using a policy that matches its name.