Add option/setting to prevent declaration of queue (#37)

Christian Häusler · web-flow · commit 5b289c79c5a4 · 2018-06-16T18:35:34.000+02:00
In certain setups, queue will already be declared when a consumer gets started. Starting the consumer will fail if the declared queue and the consumers settings mismatch. In order to prevent the need for managing the queue settings in multiple places, one might want to prevent the consumer from declaring the queue. This change adds a new cli option `-no-declare` and a new setting `nodeclare` to the `[queuesettings]` section of the config file which allows to do exactly this. resolves #36.
diff --git a/config/config.go b/config/config.go
@@ -32,6 +32,7 @@ type Config struct {
 		DeadLetterExchange   string
 		DeadLetterRoutingKey string
 		Priority             int
+		Nodeclare            bool
 	}
 	Exchange struct {
 		Name       string
@@ -76,6 +77,16 @@ func (c *Config) AmqpUrl() string {
 	return c.RabbitMq.AmqpUrl
 }
 
+// QueueName returns the name of toe queue to bind with.
+func (c Config) QueueName() string {
+	return c.RabbitMq.Queue
+}
+
+// MustDeclareQueue return if the consumer should declare the queue or if the queue is expected to be already declared.
+func (c Config) MustDeclareQueue() bool {
+	return !c.QueueSettings.Nodeclare
+}
+
 // HasExchange checks if an exchange is configured.
 func (c Config) HasExchange() bool {
 	return c.Exchange.Name != ""
@@ -96,6 +107,16 @@ func (c Config) ExchangeType() string {
 	return c.Exchange.Type
 }
 
+// ExchangeIsDurable returns whether the exchange should be durable or not.
+func (c Config) ExchangeIsDurable() bool {
+	return c.Exchange.Durable
+}
+
+// ExchangeIsAutoDelete return whether the exchange should be auto deleted or not.
+func (c Config) ExchangeIsAutoDelete() bool {
+	return c.Exchange.Autodelete
+}
+
 // PrefetchCount returns the configured prefetch count of the QoS settings.
 func (c Config) PrefetchCount() int {
 	// Attempt to preserve BC here
@@ -106,6 +127,12 @@ func (c Config) PrefetchCount() int {
 	return c.Prefetch.Count
 }
 
+// PrefetchIsGlobal returns if the prefetch count is defined globally for all consumers or locally for just each single
+// consumer.
+func (c Config) PrefetchIsGlobal() bool {
+	return c.Prefetch.Global
+}
+
 // HasMessageTTL checks if a message TTL is configured.
 func (c Config) HasMessageTTL() bool {
 	return c.QueueSettings.MessageTTL > 0
diff --git a/consumer/config.go b/consumer/config.go
@@ -0,0 +1,25 @@
+package consumer
+
+// Config defines the interface to present configurations to the consumer.
+type Config interface {
+	AmqpUrl() string
+	ConsumerTag() string
+	DeadLetterExchange() string
+	DeadLetterRoutingKey() string
+	ExchangeIsAutoDelete() bool
+	ExchangeIsDurable() bool
+	ExchangeName() string
+	ExchangeType() string
+	HasDeadLetterExchange() bool
+	HasDeadLetterRouting() bool
+	HasExchange() bool
+	HasMessageTTL() bool
+	HasPriority() bool
+	MessageTTL() int32
+	MustDeclareQueue() bool
+	PrefetchCount() int
+	PrefetchIsGlobal() bool
+	Priority() int32
+	QueueName() string
+	RoutingKeys() []string
+}
diff --git a/consumer/consumer.go b/consumer/consumer.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/corvus-ch/rabbitmq-cli-consumer/config"
 	"github.com/corvus-ch/rabbitmq-cli-consumer/delivery"
 	"github.com/corvus-ch/rabbitmq-cli-consumer/processor"
 	"github.com/streadway/amqp"
@@ -34,7 +33,7 @@ func New(conn Connection, ch Channel, p processor.Processor, l logr.Logger) *Con
 
 // NewFromConfig creates a new consumer instance. The setup of the amqp connection and channel is done according to the
 // configuration.
-func NewFromConfig(cfg *config.Config, p processor.Processor, l logr.Logger) (*Consumer, error) {
+func NewFromConfig(cfg Config, p processor.Processor, l logr.Logger) (*Consumer, error) {
 	l.Info("Connecting RabbitMQ...")
 	conn, err := amqp.Dial(cfg.AmqpUrl())
 	if nil != err {
@@ -54,7 +53,7 @@ func NewFromConfig(cfg *config.Config, p processor.Processor, l logr.Logger) (*C
 	return &Consumer{
 		Connection: conn,
 		Channel:    ch,
-		Queue:      cfg.RabbitMq.Queue,
+		Queue:      cfg.QueueName(),
 		Tag:        cfg.ConsumerTag(),
 		Processor:  p,
 		Log:        l,
diff --git a/consumer/setup.go b/consumer/setup.go
@@ -3,20 +3,22 @@ package consumer
 import (
 	"fmt"
 
-	"github.com/corvus-ch/rabbitmq-cli-consumer/config"
 	"github.com/streadway/amqp"
 	"github.com/thockin/logr"
 )
 
 // Setup configures queues, exchanges and bindings in between according to the configuration.
-func Setup(cfg *config.Config, ch Channel, l logr.Logger) error {
+func Setup(cfg Config, ch Channel, l logr.Logger) error {
 	if err := setupQoS(cfg, ch, l); err != nil {
 		return err
 	}
 
-	if err := declareQueue(cfg, ch, l); err != nil {
-		return err
+	if cfg.MustDeclareQueue() {
+		if err := declareQueue(cfg, ch, l); err != nil {
+			return err
+		}
 	}
+
 	// Empty Exchange name means default, no need to declare
 	if cfg.HasExchange() {
 		if err := declareExchange(cfg, ch, l); err != nil {
@@ -27,32 +29,32 @@ func Setup(cfg *config.Config, ch Channel, l logr.Logger) error {
 	return nil
 }
 
-func setupQoS(cfg *config.Config, ch Channel, l logr.Logger) error {
+func setupQoS(cfg Config, ch Channel, l logr.Logger) error {
 	l.Info("Setting QoS... ")
-	if err := ch.Qos(cfg.PrefetchCount(), 0, cfg.Prefetch.Global); err != nil {
+	if err := ch.Qos(cfg.PrefetchCount(), 0, cfg.PrefetchIsGlobal()); err != nil {
 		return fmt.Errorf("failed to set QoS: %v", err)
 	}
 	l.Info("Succeeded setting QoS.")
 	return nil
 }
 
-func declareQueue(cfg *config.Config, ch Channel, l logr.Logger) error {
-	l.Infof("Declaring queue \"%s\"...", cfg.RabbitMq.Queue)
-	_, err := ch.QueueDeclare(cfg.RabbitMq.Queue, true, false, false, false, queueArgs(cfg))
+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))
 	if nil != err {
 		return fmt.Errorf("failed to declare queue: %v", err)
 	}
 
 	return nil
 }
 
-func declareExchange(cfg *config.Config, ch Channel, l logr.Logger) error {
-	l.Infof("Declaring exchange \"%s\"...", cfg.Exchange.Name)
+func declareExchange(cfg Config, ch Channel, l logr.Logger) error {
+	l.Infof("Declaring exchange \"%s\"...", cfg.ExchangeName())
 	if err := ch.ExchangeDeclare(
-		cfg.Exchange.Name,
+		cfg.ExchangeName(),
 		cfg.ExchangeType(),
-		cfg.Exchange.Durable,
-		cfg.Exchange.Autodelete,
+		cfg.ExchangeIsDurable(),
+		cfg.ExchangeIsAutoDelete(),
 		false,
 		false,
 		amqp.Table{},
@@ -61,10 +63,10 @@ func declareExchange(cfg *config.Config, ch Channel, l logr.Logger) error {
 	}
 
 	// Bind queue
-	l.Infof("Binding queue \"%s\" to exchange \"%s\"...", cfg.RabbitMq.Queue, cfg.Exchange.Name)
+	l.Infof("Binding queue \"%s\" to exchange \"%s\"...", cfg.QueueName(), cfg.ExchangeName())
 	for _, routingKey := range cfg.RoutingKeys() {
 		if err := ch.QueueBind(
-			cfg.RabbitMq.Queue,
+			cfg.QueueName(),
 			routingKey,
 			cfg.ExchangeName(),
 			false,
@@ -77,7 +79,7 @@ func declareExchange(cfg *config.Config, ch Channel, l logr.Logger) error {
 	return nil
 }
 
-func queueArgs(cfg *config.Config) amqp.Table {
+func queueArgs(cfg Config) amqp.Table {
 
 	args := make(amqp.Table)
 
diff --git a/example.conf b/example.conf
@@ -103,6 +103,10 @@ deadLetterroutingkey = someroutingkey
 # The priority range for this queue.
 priority = 10
 
+# Prevents the queue from being declared. If set to true, the queue must have been configured previous to starting the
+# consumer. If the queue is not defined, the consumer can not connect and quits.
+nodeclare = false
+
 [logs]
 # Path to the log file where informational output is written to
 # #
diff --git a/fixtures/TestEndToEnd/noDeclareConfigError.golden b/fixtures/TestEndToEnd/noDeclareConfigError.golden
diff --git a/fixtures/TestEndToEnd/noDeclareConfigOutput.golden b/fixtures/TestEndToEnd/noDeclareConfigOutput.golden
@@ -0,0 +1,9 @@
+Connecting RabbitMQ...
+Connected.
+Opening channel...
+Done.
+Setting QoS... 
+Succeeded setting QoS.
+Registering consumer... 
+Succeeded registering consumer.
+Waiting for messages...
diff --git a/fixtures/TestEndToEnd/noDeclareError.golden b/fixtures/TestEndToEnd/noDeclareError.golden
diff --git a/fixtures/TestEndToEnd/noDeclareOutput.golden b/fixtures/TestEndToEnd/noDeclareOutput.golden
@@ -0,0 +1,9 @@
+Connecting RabbitMQ...
+Connected.
+Opening channel...
+Done.
+Setting QoS... 
+Succeeded setting QoS.
+Registering consumer... 
+Succeeded registering consumer.
+Waiting for messages...
diff --git a/fixtures/no_declare.conf b/fixtures/no_declare.conf
@@ -0,0 +1,7 @@
+[rabbitmq]
+host = localhost
+username = guest
+password = guest
+
+[queueSettings]
+nodeclare = true
diff --git a/integration_test.go b/integration_test.go
@@ -14,9 +14,16 @@ import (
 
 	"github.com/sebdah/goldie"
 	"github.com/streadway/amqp"
+	"github.com/stretchr/testify/assert"
 )
 
-var command = os.Args[0] + " -test.run=TestHelperProcess -- "
+var (
+	command  = os.Args[0] + " -test.run=TestHelperProcess -- "
+	amqpArgs = amqp.Table{
+		"x-message-ttl":  int32(42),
+		"x-max-priority": int32(42),
+	}
+)
 
 var tests = []struct {
 	name string
@@ -112,33 +119,63 @@ var tests = []struct {
 	},
 }
 
+var noDeclareTests = []struct {
+	name string
+	// The arguments passed to the consumer command.
+	args []string
+}{
+	{"noDeclare", []string{"-V", "-no-datetime", "-q", "noDeclare", "-e", command, "-no-declare"}},
+	{"noDeclareConfig", []string{"-V", "-no-datetime", "-q", "noDeclareConfig", "-e", command, "-c", "fixtures/no_declare.conf"}},
+}
+
 func TestEndToEnd(t *testing.T) {
-	conn := prepare(t)
+	conn, ch := prepare(t)
 	defer conn.Close()
-
-	ch, err := conn.Channel()
-	if err != nil {
-		t.Errorf("failed to open channel: %v", err)
-	}
 	defer ch.Close()
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			os.Remove("./command.log")
 			cmd, stdout, stderr := startConsumer(t, test.env, test.args...)
 			declareQueueAndPublish(t, ch, test.queue, test.msg)
-			waitMessageProcessed(t, stdout)
+			waitForOutput(t, stdout, "Processed!")
 			stopConsumer(t, cmd)
 
 			output, _ := ioutil.ReadFile("./command.log")
 			goldie.Assert(t, t.Name()+"Command", output)
-			goldie.Assert(t, t.Name()+"Output", bytes.Trim(stdout.Bytes(), "\x00"))
-			goldie.Assert(t, t.Name()+"Error", bytes.Trim(stderr.Bytes(), "\x00"))
+			assertOutput(t, stdout, stderr)
 		})
 	}
+
+	for _, test := range noDeclareTests {
+		t.Run(test.name, func(t *testing.T) {
+			declareQueue(t, ch, test.name, amqpArgs)
+
+			cmd, stdout, stderr := startConsumer(t, []string{}, test.args...)
+			waitForOutput(t, stdout, "Waiting for messages...")
+			stopConsumer(t, cmd)
+
+			assertOutput(t, stdout, stderr)
+		})
+	}
+
+	t.Run("declareError", func(t *testing.T) {
+		declareQueue(t, ch, t.Name(), amqpArgs)
+
+		cmd, _, _ := startConsumer(t, []string{}, "-V", "-no-datetime", "-q", t.Name(), "-e", command)
+		exitErr := cmd.Wait()
+
+		assert.NotNil(t, exitErr)
+		assert.Equal(t, "exit status 1", exitErr.Error())
+	})
+}
+
+func assertOutput(t *testing.T, stdout, stderr *bytes.Buffer) {
+	goldie.Assert(t, t.Name()+"Output", bytes.Trim(stdout.Bytes(), "\x00"))
+	goldie.Assert(t, t.Name()+"Error", bytes.Trim(stderr.Bytes(), "\x00"))
 }
 
-func prepare(t *testing.T) *amqp.Connection {
+func prepare(t *testing.T) (*amqp.Connection, *amqp.Channel) {
 	makeCmd := exec.Command("make", "build")
 	if err := makeCmd.Run(); err != nil {
 		t.Fatalf("could not build binary for: %v", err)
@@ -159,7 +196,12 @@ func prepare(t *testing.T) *amqp.Connection {
 		t.Fatalf("failed to open AMQP connection: %v", err)
 	}
 
-	return conn
+	ch, err := conn.Channel()
+	if err != nil {
+		t.Fatalf("failed to open channel: %v", err)
+	}
+
+	return conn, ch
 }
 
 func connect(url string) (*amqp.Connection, error) {
@@ -180,14 +222,18 @@ func connect(url string) (*amqp.Connection, error) {
 	}
 }
 
-func declareQueueAndPublish(t *testing.T, ch *amqp.Channel, name string, msg amqp.Publishing) {
-	q, err := ch.QueueDeclare(name, true, false, false, false, nil)
+func declareQueue(t *testing.T, ch *amqp.Channel, name string, args amqp.Table) amqp.Queue {
+	q, err := ch.QueueDeclare(name, true, false, false, false, args)
 	if err != nil {
 		t.Errorf("failed to declare queue; %v", err)
 	}
 
-	err = ch.Publish("", q.Name, false, false, msg)
-	if err != nil {
+	return q
+}
+
+func declareQueueAndPublish(t *testing.T, ch *amqp.Channel, name string, msg amqp.Publishing) {
+	q := declareQueue(t, ch, name, nil)
+	if err := ch.Publish("", q.Name, false, false, msg); nil != err {
 		t.Errorf("failed to publish message: %v", err)
 	}
 }
@@ -213,19 +259,19 @@ func stopConsumer(t *testing.T, cmd *exec.Cmd) {
 	}
 }
 
-func waitMessageProcessed(t *testing.T, buf *bytes.Buffer) {
+func waitForOutput(t *testing.T, buf *bytes.Buffer, expect string) {
 	timeout := time.After(10 * time.Second)
 	ticker := time.NewTicker(100 * time.Millisecond)
 	defer ticker.Stop()
 
 	for {
 		select {
 		case <-timeout:
-			t.Error("timeout while waiting for message processing")
+			t.Errorf("timeout while waiting for output \"%s\"", expect)
 			return
 
 		case <-ticker.C:
-			if strings.Contains(buf.String(), "Processed!") {
+			if strings.Contains(buf.String(), expect) {
 				return
 			}
 		}
diff --git a/main.go b/main.go
