Parse amqpDSN in order to fail early

Author: boekkooi-fresh

extension/amqp/amqp.go

@@ -36,10 +36,10 @@ func setup(url, queue string) (io.Closer, NotificationChannel, error) {
 
 // DirectQueueConsume returns a Consume func that will connect to the provided AMQP server and create a queue for direct message delivery
 func DirectQueueConsume(amqpDSN, queue string) (Consume, error) {
-	switch {
-	case len(amqpDSN) == 0:
+	if _, err := amqp.ParseURI(amqpDSN); err != nil {
 		return nil, goengine.InvalidArgumentError("amqpDSN")
-	case len(queue) == 0:
+	}
+	if len(queue) == 0 {
 		return nil, goengine.InvalidArgumentError("queue")
 	}
 
@@ -54,7 +54,7 @@ func DirectQueueConsume(amqpDSN, queue string) (Consume, error) {
 			return nil, nil, err
 		}
 
-		// Exclusive consumer
+		// Since there can be multiple consumers, fair distribution of deliveries is required
 		deliveries, err := ch.Consume(queue, "", true, false, false, false, nil)
 
 		return conn, deliveries, err

extension/amqp/amqp_test.go

@@ -1,6 +1,15 @@
+// +build unit
+
 package amqp_test
 
-import "github.com/streadway/amqp"
+import (
+	"testing"
+
+	"github.com/hellofresh/goengine"
+	goengineAmqp "github.com/hellofresh/goengine/extension/amqp"
+	"github.com/streadway/amqp"
+	"github.com/stretchr/testify/assert"
+)
 
 type mockConnection struct {
 }
@@ -36,3 +45,20 @@ func (ch mockChannel) Consume(
 func (ch mockChannel) Qos(prefetchCount, prefetchSize int, global bool) error {
 	return nil
 }
+
+func TestDirectQueueConsume(t *testing.T) {
+	t.Run("Invalid arguments", func(t *testing.T) {
+		_, err := goengineAmqp.DirectQueueConsume("http://localhost:5672/", "my-queue")
+		assert.Equal(t, goengine.InvalidArgumentError("amqpDSN"), err)
+
+		_, err = goengineAmqp.DirectQueueConsume("amqp://localhost:5672/", "")
+		assert.Equal(t, goengine.InvalidArgumentError("queue"), err)
+	})
+
+	t.Run("Returns amqp.Consume", func(t *testing.T) {
+		c, err := goengineAmqp.DirectQueueConsume("amqp://localhost:5672/", "my-queue")
+		assert.NoError(t, err)
+		assert.NotNil(t, c)
+		assert.IsType(t, (goengineAmqp.Consume)(nil), c)
+	})
+}

extension/amqp/publisher_test.go

@@ -58,10 +58,3 @@ func TestNotificationPublisher_Publish(t *testing.T) {
 		ensure.Len(loggerHook.Entries, 0)
 	})
 }
-
-//func getLogger() (goengine.Logger, *test.Hook) {
-//	logger, loggerHook := test.NewNullLogger()
-//	logger.SetLevel(logrus.DebugLevel)
-//
-//	return goengineLogger.Wrap(logger), loggerHook
-//}