Merge pull request #52 from hellofresh/feature/amqp-publisher

Add amqp publisher

Author: bilaljaved

driver/sql/projection.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/mailru/easyjson/jlexer"
+	"github.com/mailru/easyjson/jwriter"
 	"github.com/pkg/errors"
 
 	"github.com/hellofresh/goengine"
@@ -129,6 +130,16 @@ func (p *ProjectionNotification) UnmarshalEasyJSON(in *jlexer.Lexer) {
 	}
 }
 
+// MarshalEasyJSON supports easyjson.Marshaler interface
+func (p *ProjectionNotification) MarshalEasyJSON(w *jwriter.Writer) {
+	w.RawByte('{')
+	w.RawString("\"no\":")
+	w.Int64(p.No)
+	w.RawString(",\"aggregate_id\":")
+	w.String(p.AggregateID)
+	w.RawByte('}')
+}
+
 // GetProjectionStateSerialization returns a ProjectionStateSerialization based on the provided projection
 func GetProjectionStateSerialization(projection goengine.Projection) ProjectionStateSerialization {
 	if saga, ok := projection.(ProjectionStateSerialization); ok {

extension/amqp/amqp.go

@@ -0,0 +1,31 @@
+package amqp
+
+import (
+	"io"
+
+	"github.com/streadway/amqp"
+)
+
+// 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
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if _, err := ch.QueueDeclare(queue, true, false, false, false, nil); err != nil {
+		return nil, nil, err
+	}
+
+	return conn, ch, nil
+}
+
+// NotificationChannel represents a channel for notifications
+type NotificationChannel interface {
+	Publish(exchange, queue string, mandatory, immediate bool, msg amqp.Publishing) error
+}

extension/amqp/amqp_test.go

@@ -0,0 +1,23 @@
+package amqp_test
+
+import "github.com/streadway/amqp"
+
+type mockConnection struct {
+}
+
+type mockChannel struct {
+}
+
+func (cn mockConnection) Close() error {
+	return nil
+}
+
+func (ch mockChannel) Publish(
+	exchange string,
+	queue string,
+	mandatory bool,
+	immediate bool,
+	msg amqp.Publishing,
+) error {
+	return nil
+}

extension/amqp/publisher.go

@@ -0,0 +1,107 @@
+package amqp
+
+import (
+	"context"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/hellofresh/goengine"
+	"github.com/hellofresh/goengine/driver/sql"
+	"github.com/mailru/easyjson"
+	"github.com/streadway/amqp"
+)
+
+var _ sql.ProjectionTrigger = (&NotificationPublisher{}).Publish
+
+// NotificationPublisher is responsible of publishing a notification to queue
+type NotificationPublisher struct {
+	amqpDSN              string
+	queue                string
+	minReconnectInterval time.Duration
+	maxReconnectInterval time.Duration
+	logger               goengine.Logger
+
+	connection io.Closer
+	channel    NotificationChannel
+
+	mux sync.Mutex
+}
+
+// NewNotificationPublisher returns an instance of NotificationPublisher
+func NewNotificationPublisher(
+	amqpDSN,
+	queue string,
+	minReconnectInterval time.Duration,
+	maxReconnectInterval time.Duration,
+	logger goengine.Logger,
+	connection io.Closer,
+	channel NotificationChannel,
+) (*NotificationPublisher, error) {
+
+	if _, err := amqp.ParseURI(amqpDSN); err != nil {
+		return nil, goengine.InvalidArgumentError("amqpDSN")
+	}
+	if len(queue) == 0 {
+		return nil, goengine.InvalidArgumentError("queue")
+	}
+	return &NotificationPublisher{
+		amqpDSN:              amqpDSN,
+		queue:                queue,
+		minReconnectInterval: minReconnectInterval,
+		maxReconnectInterval: maxReconnectInterval,
+		logger:               logger,
+		connection:           connection,
+		channel:              channel,
+	}, nil
+}
+
+// Publish sends a ProjectionNotification to Queue
+func (p *NotificationPublisher) Publish(ctx context.Context, notification *sql.ProjectionNotification) error {
+	reconnectInterval := p.minReconnectInterval
+	// Ignore nil notifications since this is not supported
+	// Skipping as we may receive a nil notification from dispatcher for the first time
+	if notification == nil {
+		p.logger.Warn("unable to handle nil notification, skipping", nil)
+		return nil
+	}
+
+	msgBody, err := easyjson.Marshal(notification)
+	if err != nil {
+		return err
+	}
+
+	for {
+		p.mux.Lock()
+		if p.connection == nil {
+			p.connection, p.channel, err = setup(p.amqpDSN, p.queue)
+			if err != nil {
+				return err
+			}
+		}
+		p.mux.Unlock()
+
+		err = p.channel.Publish("", p.queue, true, false, amqp.Publishing{
+			Body: msgBody,
+		})
+		if err == amqp.ErrClosed || err == amqp.ErrFrame || err == amqp.ErrUnexpectedFrame {
+			if err := p.connection.Close(); err != nil {
+				p.logger.Error("failed to close amqp connection", func(entry goengine.LoggerEntry) {
+					entry.Error(err)
+				})
+			}
+
+			time.Sleep(reconnectInterval)
+			reconnectInterval *= 2
+			if reconnectInterval > p.maxReconnectInterval {
+				reconnectInterval = p.maxReconnectInterval
+			}
+
+			p.connection = nil
+			p.channel = nil
+			continue
+		}
+
+		return err
+	}
+}

extension/amqp/publisher_test.go

@@ -0,0 +1,68 @@
+package amqp_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/hellofresh/goengine"
+	"github.com/hellofresh/goengine/driver/sql"
+	goengineAmqp "github.com/hellofresh/goengine/extension/amqp"
+	goengineLogger "github.com/hellofresh/goengine/extension/logrus"
+	"github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus/hooks/test"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNotificationPublisher_Publish(t *testing.T) {
+
+	ctx, ctxCancel := context.WithTimeout(context.Background(), time.Second)
+	defer ctxCancel()
+
+	channel := &mockChannel{}
+	connection := &mockConnection{}
+
+	t.Run("Invalid arguments", func(t *testing.T) {
+		logger, _ := getLogger()
+
+		_, err := goengineAmqp.NewNotificationPublisher("http://localhost:5672/", "my-queue", 3, 4, logger, connection, channel)
+		assert.Equal(t, goengine.InvalidArgumentError("amqpDSN"), err)
+
+		_, err = goengineAmqp.NewNotificationPublisher("amqp://localhost:5672/", "", 3, 4, logger, connection, channel)
+		assert.Equal(t, goengine.InvalidArgumentError("queue"), err)
+
+	})
+
+	t.Run("Publish Nil Notification Message", func(t *testing.T) {
+		ensure := require.New(t)
+		logger, loggerHook := getLogger()
+
+		publisher, err := goengineAmqp.NewNotificationPublisher("amqp://localhost:5672/", "my-queue", 3, 4, logger, connection, channel)
+		ensure.NoError(err)
+		err = publisher.Publish(ctx, nil)
+		ensure.Nil(err)
+		ensure.Len(loggerHook.Entries, 1)
+		ensure.Equal("unable to handle nil notification, skipping", loggerHook.LastEntry().Message)
+	})
+
+	t.Run("Publish Message", func(t *testing.T) {
+		ensure := require.New(t)
+		logger, loggerHook := getLogger()
+
+		publisher, err := goengineAmqp.NewNotificationPublisher("amqp://localhost:5672/", "my-queue", 3, 4, logger, connection, channel)
+		ensure.NoError(err)
+
+		err = publisher.Publish(ctx, &sql.ProjectionNotification{No: 1, AggregateID: "8150276e-34fe-49d9-aeae-a35af0040a4f"})
+
+		ensure.NoError(err)
+		ensure.Len(loggerHook.Entries, 0)
+	})
+}
+
+func getLogger() (goengine.Logger, *test.Hook) {
+	logger, loggerHook := test.NewNullLogger()
+	logger.SetLevel(logrus.DebugLevel)
+
+	return goengineLogger.Wrap(logger), loggerHook
+}

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/pkg/errors v0.8.0
 	github.com/prometheus/client_golang v0.9.4
 	github.com/sirupsen/logrus v1.2.0
+	github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271
 	github.com/stretchr/testify v1.3.0
 	go.uber.org/atomic v1.3.2 // indirect
 	go.uber.org/multierr v1.1.0 // indirect

go.sum

@@ -48,9 +48,10 @@ github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y8
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.2 h1:6LJUbpNm42llc4HRCuvApCSWB/WfhuNo9K98Q9sNGfs=
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
-github.com/sirupsen/logrus v1.0.6/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
 github.com/sirupsen/logrus v1.2.0 h1:juTguoYk5qI21pwyTXY3B3Y5cOTH3ZUyZCg1v/mihuo=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271 h1:WhxRHzgeVGETMlmVfqhRn8RIeeNoPr2Czh33I4Zdccw=
+github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
@@ -62,13 +63,11 @@ go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/zap v1.9.1 h1:XCJQEf3W6eZaVwhRBof6ImoYGJSITeKWsyeh3HFu/5o=
 go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
-golang.org/x/crypto v0.0.0-20180808211826-de0752318171/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793 h1:u+LnwYTOOW7Ukr/fppxEb1Nwz0AtPflrblfvUudpo+I=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sys v0.0.0-20180810173357-98c5dad5d1a0/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5 h1:mzjBh+S5frKOsOBobWIMAbXavqjmgO17k/2puhcFR94=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=