Merge pull request #51 from hellofresh/amqp-listen-publish

Add Amqp listener

Author: bilaljaved

extension/amqp/amqp.go

@@ -3,11 +3,20 @@ package amqp
 import (
 	"io"
 
+	"github.com/hellofresh/goengine"
 	"github.com/streadway/amqp"
 )
 
+// NotificationChannel represents a channel for notifications
+type NotificationChannel interface {
+	Publish(exchange, queue string, mandatory, immediate bool, msg amqp.Publishing) error
+	Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error)
+	Qos(prefetchCount, prefetchSize int, global bool) error
+}
+
 // setup returns a connection and channel to be used for the Queue setup
 func setup(url, queue string) (io.Closer, NotificationChannel, error) {
+
 	conn, err := amqp.Dial(url)
 	if err != nil {
 		return nil, nil, err
@@ -25,7 +34,29 @@ func setup(url, queue string) (io.Closer, NotificationChannel, error) {
 	return conn, ch, nil
 }
 
-// NotificationChannel represents a channel for notifications
-type NotificationChannel interface {
-	Publish(exchange, queue string, mandatory, immediate bool, msg amqp.Publishing) 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) {
+	if _, err := amqp.ParseURI(amqpDSN); err != nil {
+		return nil, goengine.InvalidArgumentError("amqpDSN")
+	}
+	if len(queue) == 0 {
+		return nil, goengine.InvalidArgumentError("queue")
+	}
+
+	return func() (io.Closer, <-chan amqp.Delivery, error) {
+		conn, ch, err := setup(amqpDSN, queue)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		// Indicate we only want 1 message to be acknowledge at a time.
+		if err := ch.Qos(1, 0, false); err != nil {
+			return nil, nil, err
+		}
+
+		// Since there can be multiple consumers, fair distribution of deliveries is required
+		deliveries, err := ch.Consume(queue, "", false, false, false, false, nil)
+
+		return conn, deliveries, err
+	}, nil
 }

extension/amqp/amqp_test.go

@@ -1,6 +1,30 @@
+// +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 mockAcknowledger struct {
+}
+
+func (ack mockAcknowledger) Ack(tag uint64, multiple bool) error {
+	return nil
+}
+
+func (ack mockAcknowledger) Nack(tag uint64, multiple bool, requeue bool) error {
+	return nil
+}
+
+func (ack mockAcknowledger) Reject(tag uint64, requeue bool) error {
+	return nil
+}
 
 type mockConnection struct {
 }
@@ -21,3 +45,35 @@ func (ch mockChannel) Publish(
 ) error {
 	return nil
 }
+
+func (ch mockChannel) Consume(
+	queue,
+	consumer string,
+	autoAck,
+	exclusive,
+	noLocal,
+	noWait bool,
+	args amqp.Table,
+) (<-chan amqp.Delivery, error) {
+	return make(chan amqp.Delivery), nil
+}
+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/listener.go

@@ -0,0 +1,142 @@
+package amqp
+
+import (
+	"context"
+	"io"
+	"time"
+
+	"github.com/hellofresh/goengine"
+	"github.com/hellofresh/goengine/driver/sql"
+	"github.com/mailru/easyjson"
+	"github.com/streadway/amqp"
+)
+
+// Ensure Listener implements sql.Listener
+var _ sql.Listener = &Listener{}
+
+type (
+	// Consume returns a channel of amqp.Delivery's and a related closer or an error
+	Consume func() (io.Closer, <-chan amqp.Delivery, error)
+
+	// Listener consumes messages from an queue
+	Listener struct {
+		consume              Consume
+		minReconnectInterval time.Duration
+		maxReconnectInterval time.Duration
+		logger               goengine.Logger
+	}
+)
+
+// NewListener returns a new Listener
+func NewListener(
+	consume Consume,
+	minReconnectInterval time.Duration,
+	maxReconnectInterval time.Duration,
+	logger goengine.Logger,
+) (*Listener, error) {
+	switch {
+	case consume == nil:
+		return nil, goengine.InvalidArgumentError("consume")
+	}
+
+	if logger == nil {
+		logger = goengine.NopLogger
+	}
+
+	return &Listener{
+		consume:              consume,
+		minReconnectInterval: minReconnectInterval,
+		maxReconnectInterval: maxReconnectInterval,
+		logger:               logger,
+	}, nil
+}
+
+// Listen receives messages from a queue, transforms them into a sql.ProjectionNotification and calls the trigger
+func (l *Listener) Listen(ctx context.Context, trigger sql.ProjectionTrigger) error {
+	var nextReconnect time.Time
+	reconnectInterval := l.minReconnectInterval
+	for {
+		select {
+		case <-ctx.Done():
+			return context.Canceled
+		default:
+		}
+
+		conn, deliveries, err := l.consume()
+		if err != nil {
+			l.logger.Error("failed to start consuming amqp messages", func(entry goengine.LoggerEntry) {
+				entry.Error(err)
+				entry.String("reconnect_in", reconnectInterval.String())
+			})
+
+			time.Sleep(reconnectInterval)
+			reconnectInterval *= 2
+			if reconnectInterval > l.maxReconnectInterval {
+				reconnectInterval = l.maxReconnectInterval
+			}
+			continue
+		}
+		reconnectInterval = l.minReconnectInterval
+		nextReconnect = time.Now().Add(reconnectInterval)
+
+		l.consumeMessages(ctx, conn, deliveries, trigger)
+
+		select {
+		case <-ctx.Done():
+			return context.Canceled
+		default:
+			time.Sleep(time.Until(nextReconnect))
+		}
+	}
+}
+
+func (l *Listener) consumeMessages(ctx context.Context, conn io.Closer, deliveries <-chan amqp.Delivery, trigger sql.ProjectionTrigger) {
+	defer func() {
+		if conn == nil {
+			return
+		}
+
+		if err := conn.Close(); err != nil {
+			l.logger.Error("failed to close amqp connection", func(entry goengine.LoggerEntry) {
+				entry.Error(err)
+			})
+		}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case msg, ok := <-deliveries:
+			if !ok {
+				return
+			}
+
+			notification := &sql.ProjectionNotification{}
+			if err := easyjson.Unmarshal(msg.Body, notification); err != nil {
+				l.logger.Error("failed to unmarshal delivery, dropping message", func(entry goengine.LoggerEntry) {
+					entry.Error(err)
+				})
+				continue
+			}
+
+			if err := msg.Ack(false); err != nil {
+				l.logger.Error("failed to acknowledge notification delivery", func(entry goengine.LoggerEntry) {
+					entry.Error(err)
+					entry.Int64("notification.no", notification.No)
+					entry.String("notification.aggregate_id", notification.AggregateID)
+				})
+				continue
+			}
+
+			if err := trigger(ctx, notification); err != nil {
+				l.logger.Error("failed to project notification", func(entry goengine.LoggerEntry) {
+					entry.Error(err)
+					entry.Int64("notification.no", notification.No)
+					entry.String("notification.aggregate_id", notification.AggregateID)
+				})
+			}
+
+		}
+	}
+}