Merge pull request #5 from lorenzoranucci/add-headers-support

Add headers feature

.devenv/tor-bare-metal.yaml

@@ -4,6 +4,8 @@ dbUser: root
 dbPassword: root
 dbOutboxTableRef: my_schema.my_outbox_table
 aggregateTypeRegexp: "(?i)^order$"
+dbHeadersColumnsNames:
+  - uuid
 
 kafkaBrokers: localhost:9093
 kafkaTopic: outbox_topic

.devenv/tor-docker.yaml

@@ -4,6 +4,9 @@ dbUser: root
 dbPassword: root
 dbOutboxTableRef: my_schema.my_outbox_table
 aggregateTypeRegexp: "(?i)^order$"
+dbHeadersColumnsNames:
+  - uuid
+
 
 kafkaBrokers: kafka:9092
 kafkaTopic: outbox_topic

README.md

@@ -63,8 +63,6 @@ Tor is composed of several modules so it can be extensible and make dependencies
 
 Set up the system:
 ```shell
-cd .devenv
-
 make up
 ```
 

adapters/kafka/event_dispatcher.go

@@ -1,5 +1,7 @@
 package kafka
 
+import "github.com/Shopify/sarama"
+
 func NewEventDispatcher(producer *Producer) *EventDispatcher {
 	return &EventDispatcher{producer: producer}
 }
@@ -8,6 +10,26 @@ type EventDispatcher struct {
 	producer *Producer
 }
 
-func (k *EventDispatcher) Dispatch(routingKey string, event []byte) error {
-	return k.producer.Dispatch(routingKey, event)
+func (k *EventDispatcher) Dispatch(
+	routingKey string,
+	event []byte,
+	headers []struct {
+		Key   []byte
+		Value []byte
+	},
+) error {
+	return k.producer.Dispatch(routingKey, event, mapHeaders(headers))
+}
+
+func mapHeaders(h []struct {
+	Key   []byte
+	Value []byte
+}) []sarama.RecordHeader {
+	r := make([]sarama.RecordHeader, 0, len(h))
+
+	for _, v := range h {
+		r = append(r, v)
+	}
+
+	return r
 }

adapters/kafka/producer.go

@@ -18,12 +18,17 @@ func NewProducer(brokers []string, topic string) (*Producer, error) {
 	return &Producer{syncProducer: syncProducer, topic: topic}, nil
 }
 
-func (p *Producer) Dispatch(key string, message []byte) error {
+func (p *Producer) Dispatch(
+	key string,
+	message []byte,
+	headers []sarama.RecordHeader,
+) error {
 	_, _, err := p.syncProducer.SendMessage(
 		&sarama.ProducerMessage{
-			Key:   sarama.StringEncoder(key),
-			Topic: p.topic,
-			Value: sarama.ByteEncoder(message),
+			Key:     sarama.StringEncoder(key),
+			Topic:   p.topic,
+			Value:   sarama.ByteEncoder(message),
+			Headers: headers,
 		},
 	)
 

example/api-server/cmd/create_outbox_table.sql

@@ -1,6 +1,7 @@
 CREATE TABLE IF NOT EXISTS my_schema.my_outbox_table
 (
     id             BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
+    uuid           CHAR(36)     NOT NULL UNIQUE,
     aggregate_type VARCHAR(255) NOT NULL,
     aggregate_id   VARCHAR(255) NOT NULL,
     payload        LONGBLOB     NOT NULL

example/api-server/cmd/run.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 
 	_ "github.com/go-sql-driver/mysql"
+	"github.com/google/uuid"
 	"github.com/jmoiron/sqlx"
 	"github.com/julienschmidt/httprouter"
 	"github.com/sirupsen/logrus"
@@ -135,12 +136,13 @@ func (h *HTTPHandler) insertEvent(tx *sqlx.Tx, eventName string, orderUUID strin
 	}
 
 	query := `
-INSERT INTO my_schema.my_outbox_table (aggregate_type, aggregate_id, payload)
-VALUES ('order', :aggregate_id, :payload);`
+INSERT INTO my_schema.my_outbox_table (uuid, aggregate_type, aggregate_id, payload)
+VALUES (:uuid, 'order', :aggregate_id, :payload);`
 
 	if _, err := tx.NamedExec(
 		h.db.Rebind(query),
 		map[string]interface{}{
+			"uuid":         uuid.New().String(),
 			"aggregate_id": orderUUID,
 			"payload":      payload,
 		},

example/api-server/go.mod

@@ -4,6 +4,7 @@ go 1.19
 
 require (
 	github.com/go-sql-driver/mysql v1.6.0
+	github.com/google/uuid v1.1.2
 	github.com/jmoiron/sqlx v1.3.5
 	github.com/julienschmidt/httprouter v1.3.0
 	github.com/sirupsen/logrus v1.9.0

example/api-server/go.sum

@@ -115,6 +115,7 @@ github.com/google/pprof v0.0.0-20201023163331-3e6fc7fc9c4c/go.mod h1:kpwsk12EmLe
 github.com/google/pprof v0.0.0-20201203190320-1bf35d6f28c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/pprof v0.0.0-20201218002935-b9804c9f04c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
+github.com/google/uuid v1.1.2 h1:EVhdT+1Kseyi1/pUmXKaFxYsDNy9RQYkMWRH68J/W7Y=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=

example/tor/cmd/run.go

@@ -45,6 +45,7 @@ var runCmd = &cobra.Command{
 			viper.GetString("dbAggregateIDColumnName"),
 			viper.GetString("dbAggregateTypeColumnName"),
 			viper.GetString("dbPayloadColumnName"),
+			viper.GetStringSlice("dbHeadersColumnsNames"),
 			aggregateTypeRegexp,
 		)
 		if err != nil {
@@ -66,6 +67,7 @@ func init() {
 	viper.MustBindEnv("dbAggregateIDColumnName", "DB_AGGREGATE_ID_COLUMN_NAME")
 	viper.MustBindEnv("dbAggregateTypeColumnName", "DB_AGGREGATE_TYPE_COLUMN_NAME")
 	viper.MustBindEnv("dbPayloadColumnName", "DB_PAYLOAD_COLUMN_NAME")
+	viper.MustBindEnv("dbHeadersColumnsNames", "DB_HEADERS_COLUMNS_NAME")
 	viper.MustBindEnv("aggregateTypeRegexp", "AGGREGATE_TYPE_REGEXP_EXPRESSION")
 
 	viper.MustBindEnv("kafkaBrokers", "KAFKA_BROKERS")

router/pkg/run/event_handler.go

@@ -22,20 +22,34 @@ type StateHandler interface {
 	SetLastPosition(position mysql.Position) error
 }
 
-type OutboxEvent struct {
+type outboxEvent struct {
 	AggregateID string
 	Payload     []byte
+	Headers     []eventHeader
+}
+
+type eventHeader struct {
+	Key   []byte
+	Value []byte
 }
 
 type EventDispatcher interface {
-	Dispatch(routingKey string, event []byte) error
+	Dispatch(
+		routingKey string,
+		event []byte,
+		headers []struct {
+			Key   []byte
+			Value []byte
+		},
+	) error
 }
 
 func NewEventHandler(
 	eventDispatcher EventDispatcher,
 	aggregateIDColumnName string,
 	aggregateTypeColumnName string,
 	payloadColumnName string,
+	headersColumnsNames []string,
 	aggregateTypeRegexp *regexp.Regexp,
 ) (*EventHandler, error) {
 	actualAggregateIDColumnName := defaultAggregateIDColumnName
@@ -63,6 +77,7 @@ func NewEventHandler(
 			aggregateIDColumnName:   actualAggregateIDColumnName,
 			aggregateTypeColumnName: actualAggregateTypeColumnName,
 			payloadColumnName:       actualPayloadColumnName,
+			headersColumnsNames:     headersColumnsNames,
 			aggregateTypeRegexp:     actualAggregateTypeRegexp,
 		},
 		eventDispatcher: eventDispatcher,
@@ -90,7 +105,7 @@ func (h *EventHandler) OnRow(e *canal.RowsEvent) error {
 	}
 
 	for _, oe := range oes {
-		err = h.eventDispatcher.Dispatch(oe.AggregateID, oe.Payload)
+		err = h.eventDispatcher.Dispatch(oe.AggregateID, oe.Payload, mapHeaders(oe.Headers))
 		if err != nil {
 			return err
 		}
@@ -110,3 +125,23 @@ func (h *EventHandler) OnPosSynced(p mysql.Position, g mysql.GTIDSet, f bool) er
 func (h *EventHandler) String() string {
 	return "EventHandler"
 }
+
+func mapHeaders(h []eventHeader) []struct {
+	Key   []byte
+	Value []byte
+} {
+	if len(h) == 0 {
+		return nil
+	}
+
+	r := make([]struct {
+		Key   []byte
+		Value []byte
+	}, 0, len(h))
+
+	for _, v := range h {
+		r = append(r, v)
+	}
+
+	return r
+}