Add Watermill example for publishing events

_examples/postcard/cmd/main.go

@@ -0,0 +1,133 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log"
+
+	"github.com/ThreeDotsLabs/watermill"
+	"github.com/ThreeDotsLabs/watermill-amqp/pkg/amqp"
+	sql2 "github.com/ThreeDotsLabs/watermill-sql/pkg/sql"
+	"github.com/ThreeDotsLabs/watermill/components/forwarder"
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/google/uuid"
+	_ "github.com/lib/pq"
+
+	"postcard"
+	"postcard/storage"
+)
+
+func main() {
+	conn := "host=localhost port=5432 user=postgres password=password dbname=postgres sslmode=disable"
+	db, err := sql.Open("postgres", conn)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	repo, err := storage.NewDefaultSimplePostcardRepository(context.Background(), db)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	logger := watermill.NewStdLogger(false, false)
+
+	router, err := message.NewRouter(message.RouterConfig{}, logger)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	sqlSub, err := sql2.NewSubscriber(db, sql2.SubscriberConfig{
+		SchemaAdapter:    storage.WatermillSchemaAdapter{},
+		OffsetsAdapter:   sql2.DefaultPostgreSQLOffsetsAdapter{},
+		InitializeSchema: true,
+	}, logger)
+
+	amqpURI := "amqp://guest:guest@localhost:5672/"
+	amqpConfig := amqp.NewDurablePubSubConfig(amqpURI, func(topic string) string {
+		return topic
+	})
+
+	amqpPub, err := amqp.NewPublisher(amqpConfig, logger)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	amqpSub, err := amqp.NewSubscriber(amqpConfig, logger)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	router.AddHandler(
+		"sql-to-amqp",
+		"events",
+		sqlSub,
+		"events",
+		amqpPub,
+		func(msg *message.Message) ([]*message.Message, error) {
+			return []*message.Message{msg}, nil
+		},
+	)
+
+	router.AddNoPublisherHandler(
+		"read-amqp",
+		"events",
+		amqpSub,
+		func(msg *message.Message) error {
+			fmt.Printf("Received message %v: %v\n", msg.UUID, string(msg.Payload))
+			return nil
+		},
+	)
+
+	fwd, err := forwarder.NewForwarder(sqlSub, amqpPub, logger, forwarder.Config{
+		ForwarderTopic: "events",
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	go func() {
+		err := fwd.Run(context.Background())
+		if err != nil {
+			log.Fatal(err)
+		}
+	}()
+
+	go func() {
+		err = router.Run(context.Background())
+		if err != nil {
+			log.Fatal(err)
+		}
+	}()
+
+	<-fwd.Running()
+	<-router.Running()
+
+	pc, err := postcard.NewPostcard(uuid.NewString())
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = repo.Save(context.Background(), pc)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = pc.Address(postcard.Address{
+		Name:  "Someone",
+		Line1: "Somewhere",
+	}, postcard.Address{
+		Name:  "Who",
+		Line1: "Where",
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	err = repo.Save(context.Background(), pc)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	select {}
+}

_examples/postcard/docker-compose.yml

@@ -0,0 +1,14 @@
+version: '3'
+services:
+  postgres:
+    image: postgres
+    environment:
+      - POSTGRES_PASSWORD=password
+      - POSTGRES_DB=esja
+    ports:
+      - "5432:5432"
+
+  rabbitmq:
+    image: rabbitmq:3.7-management
+    ports:
+      - 5672:5672

_examples/postcard/go.mod

@@ -4,16 +4,26 @@ go 1.18
 
 require (
 	github.com/ThreeDotsLabs/esja v0.0.0-20221208191400-8fbb493947e7
+	github.com/ThreeDotsLabs/pii v0.0.0-20230103125711-e0908da9a963
+	github.com/ThreeDotsLabs/watermill v1.2.0-rc.11
+	github.com/ThreeDotsLabs/watermill-amqp v1.1.4
+	github.com/ThreeDotsLabs/watermill-sql v1.3.7
 	github.com/google/uuid v1.3.0
 	github.com/lib/pq v1.10.6
 	github.com/mattn/go-sqlite3 v1.14.16
 	github.com/stretchr/testify v1.8.1
 )
 
 require (
-	github.com/ThreeDotsLabs/pii v0.0.0-20230103125711-e0908da9a963 // indirect
+	github.com/cenkalti/backoff/v3 v3.2.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/lithammer/shortuuid/v3 v3.0.7 // indirect
+	github.com/oklog/ulid v1.3.1 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/streadway/amqp v1.0.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )