Allow to set queue declare options in config file (#50)

Vodzo · Christian Häusler · commit 2e639a6df0a7 · 2018-11-10T10:21:23.000+01:00
Resolves #49.
diff --git a/config/config.go b/config/config.go
@@ -33,6 +33,10 @@ type Config struct {
 		DeadLetterRoutingKey string
 		Priority             int
 		Nodeclare            bool
+		Durable              bool
+		Exclusive            bool
+		AutoDelete           bool
+		NoWait               bool
 	}
 	Exchange struct {
 		Name       string
@@ -192,6 +196,27 @@ func (c Config) WithDateTime() bool {
 	return !c.Logs.NoDateTime
 }
 
+// QueueIsDurable checks if queue should be declared durable.
+// Defaults to true to keep backwards compatibility
+func (c Config) QueueIsDurable() bool {
+	return c.QueueSettings.Durable
+}
+
+// QueueIsExclusive checks if queue should be declared exclusive
+func (c Config) QueueIsExclusive() bool {
+	return c.QueueSettings.Exclusive
+}
+
+// QueueIsAutoDelete checks if queue should be declared 'autoDelete'
+func (c Config) QueueIsAutoDelete() bool {
+	return c.QueueSettings.AutoDelete
+}
+
+// QueueIsNoWait checks if queue should be declared 'noWait'
+func (c Config) QueueIsNoWait() bool {
+	return c.QueueSettings.NoWait
+}
+
 // ConsumerTag returns the tag used to identify the consumer.
 func (c Config) ConsumerTag() string {
 	if v, set := os.LookupEnv("GO_WANT_HELPER_PROCESS"); set && v == "1" {
@@ -217,23 +242,34 @@ func LoadAndParse(location string) (*Config, error) {
 		location = location
 	}
 
-	cfg := Config{}
-	if err := gcfg.ReadFileInto(&cfg, location); err != nil {
+	cfg := &Config{}
+
+	SetDefaultQueueDurability(cfg)
+
+	if err := gcfg.ReadFileInto(cfg, location); err != nil {
 		return nil, err
 	}
 
-	return &cfg, nil
+	return cfg, nil
 }
 
 func CreateFromString(data string) (*Config, error) {
 	cfg := &Config{}
+
+	SetDefaultQueueDurability(cfg)
+
 	if err := gcfg.ReadStringInto(cfg, data); err != nil {
 		return nil, err
 	}
 
 	return cfg, nil
 }
 
+// SetDefaultQueueDurability sets queue durable to true to keep backwards compatibility
+func SetDefaultQueueDurability(cfg *Config) {
+	cfg.QueueSettings.Durable = true
+}
+
 func transformToStringValue(val string) string {
 	if val == "<empty>" {
 		return ""
diff --git a/consumer/config.go b/consumer/config.go
@@ -22,4 +22,8 @@ type Config interface {
 	Priority() int32
 	QueueName() string
 	RoutingKeys() []string
+	QueueIsDurable() bool
+	QueueIsExclusive() bool
+	QueueIsAutoDelete() bool
+	QueueIsNoWait() bool
 }
diff --git a/consumer/fixtures/autodelete_queue.conf b/consumer/fixtures/autodelete_queue.conf
@@ -0,0 +1,5 @@
+[rabbitmq]
+queue = autoDeleteQueue
+
+[queuesettings]
+autodelete = On
diff --git a/consumer/fixtures/default_queue_durability.conf b/consumer/fixtures/default_queue_durability.conf
@@ -0,0 +1,2 @@
+[rabbitmq]
+queue = defaultQueueDurability
diff --git a/consumer/fixtures/durable_queue.conf b/consumer/fixtures/durable_queue.conf
@@ -0,0 +1,5 @@
+[rabbitmq]
+queue = durableQueue
+
+[queuesettings]
+durable = On
diff --git a/consumer/fixtures/exclusive_queue.conf b/consumer/fixtures/exclusive_queue.conf
@@ -0,0 +1,5 @@
+[rabbitmq]
+queue = exclusiveQueue
+
+[queuesettings]
+exclusive = On
diff --git a/consumer/fixtures/non_durable_queue.conf b/consumer/fixtures/non_durable_queue.conf
@@ -0,0 +1,5 @@
+[rabbitmq]
+queue = nonDurableQueue
+
+[queuesettings]
+durable = Off
diff --git a/consumer/fixtures/nowait_queue.conf b/consumer/fixtures/nowait_queue.conf
@@ -0,0 +1,5 @@
+[rabbitmq]
+queue = noWaitQueue
+
+[queuesettings]
+nowait = On
diff --git a/consumer/setup.go b/consumer/setup.go
@@ -39,7 +39,14 @@ func setupQoS(cfg Config, ch Channel, l logr.Logger) error {
 
 func declareQueue(cfg Config, ch Channel, l logr.Logger) error {
 	l.Infof("Declaring queue \"%s\"...", cfg.QueueName())
-	_, err := ch.QueueDeclare(cfg.QueueName(), true, false, false, false, queueArgs(cfg))
+	_, err := ch.QueueDeclare(
+		cfg.QueueName(),         // Queue name
+		cfg.QueueIsDurable(),    // durable
+		cfg.QueueIsAutoDelete(), // autoDelete
+		cfg.QueueIsExclusive(),  // exclusive
+		cfg.QueueIsNoWait(),     // noWait
+		queueArgs(cfg),          // arguments
+	)
 	if nil != err {
 		if amqpErr, ok := err.(*amqp.Error); ok && amqpErr.Code == 406 {
 			l.Error("Queue already declared with conflicting settings. You might want to use --no-declare.")
diff --git a/consumer/setup_test.go b/consumer/setup_test.go
@@ -23,6 +23,12 @@ const (
 	routingConfig             = "routing"
 	simpleExchangeConfig      = "exchange"
 	ttlConfig                 = "ttl"
+	autoDeleteQueue           = "autodelete_queue"
+	durableQueue              = "durable_queue"
+	nonDurableQueue           = "non_durable_queue"
+	defaultQueueDurability    = "default_queue_durability"
+	exclusiveQueue            = "exclusive_queue"
+	noWaitQueue               = "nowait_queue"
 )
 
 var nilAmqpTable amqp.Table
@@ -203,6 +209,66 @@ var queueTests = []struct {
 		},
 		fmt.Errorf("failed to bind queue to exchange: queue bind error"),
 	},
+	// Durable queue
+	{
+		"durableQueue",
+		durableQueue,
+		func(ch *TestChannel) {
+			ch.On("Qos", 3, 0, false).Return(nil).Once()
+			ch.On("QueueDeclare", "durableQueue", true, false, false, false, emptyAmqpTable).Return(amqp.Queue{}, nil).Once()
+		},
+		nil,
+	},
+	// Non durable queue
+	{
+		"nonDurableQueue",
+		nonDurableQueue,
+		func(ch *TestChannel) {
+			ch.On("Qos", 3, 0, false).Return(nil).Once()
+			ch.On("QueueDeclare", "nonDurableQueue", false, false, false, false, emptyAmqpTable).Return(amqp.Queue{}, nil).Once()
+		},
+		nil,
+	},
+	// Default queue durability
+	{
+		"defaultQueueDurability",
+		defaultQueueDurability,
+		func(ch *TestChannel) {
+			ch.On("Qos", 3, 0, false).Return(nil).Once()
+			ch.On("QueueDeclare", "defaultQueueDurability", true, false, false, false, emptyAmqpTable).Return(amqp.Queue{}, nil).Once()
+		},
+		nil,
+	},
+	// AutoDelete queue
+	{
+		"autoDeleteQueue",
+		autoDeleteQueue,
+		func(ch *TestChannel) {
+			ch.On("Qos", 3, 0, false).Return(nil).Once()
+			ch.On("QueueDeclare", "autoDeleteQueue", true, true, false, false, emptyAmqpTable).Return(amqp.Queue{}, nil).Once()
+		},
+		nil,
+	},
+	// Exclusive queue
+	{
+		"exclusiveQueue",
+		exclusiveQueue,
+		func(ch *TestChannel) {
+			ch.On("Qos", 3, 0, false).Return(nil).Once()
+			ch.On("QueueDeclare", "exclusiveQueue", true, false, true, false, emptyAmqpTable).Return(amqp.Queue{}, nil).Once()
+		},
+		nil,
+	},
+	// Nowait queue
+	{
+		"noWaitQueue",
+		noWaitQueue,
+		func(ch *TestChannel) {
+			ch.On("Qos", 3, 0, false).Return(nil).Once()
+			ch.On("QueueDeclare", "noWaitQueue", true, false, false, true, emptyAmqpTable).Return(amqp.Queue{}, nil).Once()
+		},
+		nil,
+	},
 }
 
 func TestQueueSettings(t *testing.T) {
diff --git a/example.conf b/example.conf
@@ -107,6 +107,25 @@ priority = 10
 # consumer. If the queue is not defined, the consumer can not connect and quits.
 nodeclare = false
 
+# Should the queue be declared as durable. If set to true, the queue will survive server restarts.
+# Defaults to true.
+durable = true
+
+# Should the queue be declared as exclusive. If set to true, the queue will be removed when consumer disconnects.
+# Defaults to false.
+exclusive = false
+
+# Should the queue be declared as autodelete. If set to true and is durable, the queue will survive server restarts
+# but will be removed when there are no remaining consumers or bindings.
+# Defaults to false
+autodelete = false
+
+# Should the queue be declared as noWait. When noWait is true, the queue will assume to be declared on the server.
+# A channel exception will arrive if the conditions are met for existing queues
+# or attempting to modify an existing queue from a different connection.
+# Defaults to false
+nowait = false
+
 [logs]
 # Path to the log file where informational output is written to
 # When providing the --verbose, -V option, this section becomes optional.

Original file line number	Diff line number	Diff line change
`@@ -22,4 +22,8 @@ type Config interface {`
`22`	`22`	`Priority() int32`
`23`	`23`	`QueueName() string`
`24`	`24`	`RoutingKeys() []string`
	`25`	`+ QueueIsDurable() bool`
	`26`	`+ QueueIsExclusive() bool`
	`27`	`+ QueueIsAutoDelete() bool`
	`28`	`+ QueueIsNoWait() bool`
`25`	`29`	`}`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+[rabbitmq]`
	`2`	`+queue = defaultQueueDurability`