Add amqp publisher

Author: bilaljaved

driver/sql/projection.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/hellofresh/goengine"
 	"github.com/mailru/easyjson/jlexer"
+	"github.com/mailru/easyjson/jwriter"
 	"github.com/pkg/errors"
 )
 
@@ -128,6 +129,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,24 @@
+package amqp
+
+import (
+	"github.com/streadway/amqp"
+)
+
+// setup returns a connection and channel to be used for consumption with the Queue setup
+func setup(url, queue string) (*amqp.Connection, *amqp.Channel, 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
+}

extension/amqp/publisher.go

@@ -0,0 +1,70 @@
+package amqp
+
+import (
+	"context"
+
+	"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
+	logger  goengine.Logger
+
+	connection *amqp.Connection
+	channel    *amqp.Channel
+}
+
+// NewNotificationPublisher returns an instance of NotificationPublisher
+func NewNotificationPublisher(amqpDSN, queue string, logger goengine.Logger) (*NotificationPublisher, error) {
+	return &NotificationPublisher{
+		amqpDSN: amqpDSN,
+		queue:   queue,
+		logger:  logger,
+	}, nil
+}
+
+// Publish sends a ProjectionNotification to Queue
+func (p *NotificationPublisher) Publish(ctx context.Context, notification *sql.ProjectionNotification) error {
+	// Ignore nil notifications since this is not supported
+	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 {
+		if p.connection == nil {
+			p.connection, p.channel, err = setup(p.amqpDSN, p.queue)
+			if err != nil {
+				return err
+			}
+		}
+
+		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)
+				})
+			}
+			p.connection = nil
+			p.channel = nil
+			continue
+		}
+
+		return err
+	}
+}

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=