Add AMQP queue declaration configuration options

v1/brokers/amqp/amqp.go

@@ -62,7 +62,7 @@ func (b *Broker) StartConsuming(consumerTag string, concurrency int, taskProcess
 		false,                           // queue delete when unused
 		b.GetConfig().AMQP.BindingKey,   // queue binding key
 		nil,                             // exchange declare args
-		nil,                             // queue declare args
+		amqp.Table(b.GetConfig().AMQP.QueueDeclareArgs), // queue declare args
 		amqp.Table(b.GetConfig().AMQP.QueueBindingArgs), // queue binding args
 	)
 	if err != nil {
@@ -210,7 +210,7 @@ func (b *Broker) Publish(ctx context.Context, signature *tasks.Signature) error
 		queue,
 		bindingKey, // queue binding key
 		nil,        // exchange declare args
-		nil,        // queue declare args
+		amqp.Table(b.GetConfig().AMQP.QueueDeclareArgs), // queue declare args
 		amqp.Table(b.GetConfig().AMQP.QueueBindingArgs), // queue binding args
 	)
 	if err != nil {
@@ -229,6 +229,7 @@ func (b *Broker) Publish(ctx context.Context, signature *tasks.Signature) error
 			Headers:      amqp.Table(signature.Headers),
 			ContentType:  "application/json",
 			Body:         msg,
+			Priority:     signature.Priority,
 			DeliveryMode: amqp.Persistent,
 		},
 	); err != nil {

v1/config/config.go

@@ -70,10 +70,14 @@ type Config struct {
 // QueueBindingArgs arguments which are used when binding to the exchange
 type QueueBindingArgs map[string]interface{}
 
+// QueueDeclareArgs arguments which are used when declaring a queue
+type QueueDeclareArgs map[string]interface{}
+
 // AMQPConfig wraps RabbitMQ related configuration
 type AMQPConfig struct {
 	Exchange         string           `yaml:"exchange" envconfig:"AMQP_EXCHANGE"`
 	ExchangeType     string           `yaml:"exchange_type" envconfig:"AMQP_EXCHANGE_TYPE"`
+	QueueDeclareArgs QueueDeclareArgs `yaml:"queue_declare_args" envconfig:"AMQP_QUEUE_DECLARE_ARGS"`
 	QueueBindingArgs QueueBindingArgs `yaml:"queue_binding_args" envconfig:"AMQP_QUEUE_BINDING_ARGS"`
 	BindingKey       string           `yaml:"binding_key" envconfig:"AMQP_BINDING_KEY"`
 	PrefetchCount    int              `yaml:"prefetch_count" envconfig:"AMQP_PREFETCH_COUNT"`

v1/config/file_test.go

@@ -19,6 +19,8 @@ amqp:
   exchange: exchange
   exchange_type: exchange_type
   prefetch_count: 123
+  queue_declare_args:
+    x-max-priority: 10
   queue_binding_args:
     image-type: png
     x-match: any
@@ -55,6 +57,7 @@ func TestNewFromYaml(t *testing.T) {
 	assert.Equal(t, "exchange", cnf.AMQP.Exchange)
 	assert.Equal(t, "exchange_type", cnf.AMQP.ExchangeType)
 	assert.Equal(t, "binding_key", cnf.AMQP.BindingKey)
+	assert.Equal(t, 10, cnf.AMQP.QueueDeclareArgs["x-max-priority"])
 	assert.Equal(t, "any", cnf.AMQP.QueueBindingArgs["x-match"])
 	assert.Equal(t, "png", cnf.AMQP.QueueBindingArgs["image-type"])
 	assert.Equal(t, 123, cnf.AMQP.PrefetchCount)

v1/config/testconfig.yml

@@ -8,6 +8,8 @@ amqp:
   exchange: exchange
   exchange_type: exchange_type
   prefetch_count: 123
+  queue_declare_args:
+    x-max-priority: 10
   queue_binding_args:
     image-type: png
     x-match: any

v1/tasks/signature.go

@@ -51,6 +51,7 @@ type Signature struct {
 	GroupTaskCount int
 	Args           []Arg
 	Headers        Headers
+	Priority       uint8
 	Immutable      bool
 	RetryCount     int
 	RetryTimeout   int