From cfdafe8b0c7ecdb38628c4d467ddbab58be54583 Mon Sep 17 00:00:00 2001 From: Robert Laszczak Date: Wed, 12 Dec 2018 09:31:41 +0100 Subject: [PATCH] https://watermill.io/ docs (#18) * added docs draft * fixed netlify config * moved netlify.toml to proper dir * added extra hugo config for netlify * fix netlify configs * fixed build command dir * added support page * added snippets support * added config snippets * [docs] fixed examples * updated build script to load snippets * fixed syntax coloring * added serializer sections for pub/sub implementations * fixed code style * [docs] getting started update * [docs] added basic router info in getting started * [docs] added router example to getting started * [StdLoggerAdapter] fixed error logging * [docs] added what is watermill * [docs] simplified getting started router example * [docs] added printMessage to snippet in getting started * [docs][getting-started] added examples link * [docs] added google analytics * [docs] added more docs * [docs] added missing docs * [docs] added some links * [docs] linking snippets to line in github * [godoc] added watermill.io info * [docs] added link to getting started in implenting pub/sub todos * [examples] added docker-compose for simple-app * [docs] added router schema * [docs] added macOS librdkafka install instructions * [docs] update grammer * [docs] some cosmetics * fix hugo build * fix netlify config * added hugo version at build * [netlify] fix printing hugo version * [netlify] fixed env * [netlify] fixed hugo version --- .gitignore | 2 + README.md | 6 +- _examples/simple-app/README.md | 15 ++ _examples/simple-app/docker-compose.yml | 42 ++++ _examples/simple-app/publishing-app/go.mod | 7 + _examples/simple-app/publishing-app/go.sum | 22 ++ _examples/simple-app/publishing-app/main.go | 53 ++-- .../simple-app/subscribing-app/counter.go | 52 ---- .../subscribing-app/feed_updater.go | 44 ---- _examples/simple-app/subscribing-app/go.mod | 7 + _examples/simple-app/subscribing-app/go.sum | 36 +++ _examples/simple-app/subscribing-app/main.go | 124 ++++++++- .../simple-app/subscribing-app/metrics.go | 42 ---- doc.go | 13 + docs/archetypes/default.md | 6 + docs/archetypes/docs.md | 9 + docs/build.sh | 38 +++ docs/config.toml | 35 +++ docs/content/docs/getting-started.md | 238 ++++++++++++++++++ .../docs/getting-started/go-channel/main.go | 50 ++++ .../getting-started/kafka/docker-compose.yml | 28 +++ .../content/docs/getting-started/kafka/go.mod | 3 + .../content/docs/getting-started/kafka/go.sum | 17 ++ .../docs/getting-started/kafka/main.go | 60 +++++ .../docs/getting-started/router/main.go | 127 ++++++++++ docs/content/docs/message.md | 42 ++++ docs/content/docs/message/receiving-ack.go | 24 ++ docs/content/docs/messages-router.md | 106 ++++++++ docs/content/docs/pub-sub-implementations.md | 175 +++++++++++++ docs/content/docs/pub-sub-implementing.md | 39 +++ docs/content/docs/pub-sub.md | 80 ++++++ docs/content/docs/troubleshooting.md | 66 +++++ docs/content/support.md | 14 ++ docs/layouts/index.html | 65 +++++ docs/layouts/partials/footer.html | 14 ++ docs/layouts/shortcodes/collapse-box.html | 3 + docs/layouts/shortcodes/collapse-toggle.html | 1 + docs/layouts/shortcodes/collapse.html | 3 + docs/layouts/shortcodes/contact-email.html | 3 + docs/layouts/shortcodes/highlight.html | 1 + .../shortcodes/load-snippet-partial.html | 67 +++++ docs/layouts/shortcodes/load-snippet.html | 26 ++ docs/layouts/shortcodes/message.html | 4 + docs/layouts/shortcodes/render-md.html | 11 + docs/layouts/shortcodes/tabs-tab.html | 3 + docs/layouts/shortcodes/tabs.html | 13 + docs/static/css/custom.css | 24 ++ docs/static/img/watermill-router.svg | 1 + log.go | 4 +- message/infrastructure/gochannel/pubsub.go | 25 +- message/infrastructure/http/subscriber.go | 22 +- message/infrastructure/kafka/marshaler.go | 6 +- message/infrastructure/kafka/publisher.go | 6 +- message/infrastructure/kafka/subscriber.go | 15 +- message/message.go | 64 ++++- message/publisher.go | 8 + message/pubsub.go | 8 +- message/router.go | 59 ++++- message/subscriber.go | 7 + netlify.toml | 17 ++ 60 files changed, 1900 insertions(+), 202 deletions(-) create mode 100644 _examples/simple-app/README.md create mode 100644 _examples/simple-app/docker-compose.yml create mode 100644 _examples/simple-app/publishing-app/go.mod create mode 100644 _examples/simple-app/publishing-app/go.sum delete mode 100644 _examples/simple-app/subscribing-app/counter.go delete mode 100644 _examples/simple-app/subscribing-app/feed_updater.go create mode 100644 _examples/simple-app/subscribing-app/go.mod create mode 100644 _examples/simple-app/subscribing-app/go.sum delete mode 100644 _examples/simple-app/subscribing-app/metrics.go create mode 100644 doc.go create mode 100644 docs/archetypes/default.md create mode 100644 docs/archetypes/docs.md create mode 100755 docs/build.sh create mode 100644 docs/config.toml create mode 100644 docs/content/docs/getting-started.md create mode 100644 docs/content/docs/getting-started/go-channel/main.go create mode 100644 docs/content/docs/getting-started/kafka/docker-compose.yml create mode 100644 docs/content/docs/getting-started/kafka/go.mod create mode 100644 docs/content/docs/getting-started/kafka/go.sum create mode 100644 docs/content/docs/getting-started/kafka/main.go create mode 100644 docs/content/docs/getting-started/router/main.go create mode 100644 docs/content/docs/message.md create mode 100644 docs/content/docs/message/receiving-ack.go create mode 100644 docs/content/docs/messages-router.md create mode 100644 docs/content/docs/pub-sub-implementations.md create mode 100644 docs/content/docs/pub-sub-implementing.md create mode 100644 docs/content/docs/pub-sub.md create mode 100644 docs/content/docs/troubleshooting.md create mode 100644 docs/content/support.md create mode 100644 docs/layouts/index.html create mode 100644 docs/layouts/partials/footer.html create mode 100644 docs/layouts/shortcodes/collapse-box.html create mode 100644 docs/layouts/shortcodes/collapse-toggle.html create mode 100644 docs/layouts/shortcodes/collapse.html create mode 100644 docs/layouts/shortcodes/contact-email.html create mode 100644 docs/layouts/shortcodes/highlight.html create mode 100644 docs/layouts/shortcodes/load-snippet-partial.html create mode 100644 docs/layouts/shortcodes/load-snippet.html create mode 100644 docs/layouts/shortcodes/message.html create mode 100644 docs/layouts/shortcodes/render-md.html create mode 100644 docs/layouts/shortcodes/tabs-tab.html create mode 100644 docs/layouts/shortcodes/tabs.html create mode 100644 docs/static/css/custom.css create mode 100644 docs/static/img/watermill-router.svg create mode 100644 netlify.toml diff --git a/.gitignore b/.gitignore index 3ce5adbbd..db26ada4b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,4 @@ .idea vendor +docs/themes/ +docs/public/ diff --git a/README.md b/README.md index 995b78cfb..0b57ba130 100644 --- a/README.md +++ b/README.md @@ -37,8 +37,8 @@ You can find more about our motivations in our [*Introducing Watermill* blog pos * **Easy** to understand (see examples below). * **Universal** - event-driven architecture, messaging, stream processing, CQRS - use it for whatever you need. * **Fast** - *(benchmarks coming soon)* -* **Extendable** with middlewares and plugins. -* **Resillient** - using proven technologies and passing stress tests *(results coming soon)*. +* **Flexible** with middlewares and plugins. +* **Resilient** - using proven technologies and passing stress tests *(results coming soon)*. ## Pub/Subs @@ -92,7 +92,7 @@ and submit your pull request via GitHub. ## Support -Join us on the `#watermill` channel on the [Gophers slack](https://gophers.slack.com/): https://gophersinvite.herokuapp.com/ +Please join us on the `#watermill` channel on the [Gophers slack](https://gophers.slack.com/): You can get invite [here](https://gophersinvite.herokuapp.com/). ## Why the name? diff --git a/_examples/simple-app/README.md b/_examples/simple-app/README.md new file mode 100644 index 000000000..ff6f7a0eb --- /dev/null +++ b/_examples/simple-app/README.md @@ -0,0 +1,15 @@ +# Simple app + +This app has separated publisher - [`publishing-app`](publishing-app/) and [`subscribing-app`](subscribing-app/). + +Warning: Subscribing app has throttling middleware enabled. + +## Requirements + +To run this example you will need Docker and docker-compose installed. See installation guide at https://docs.docker.com/compose/install/ + +## Running + +```bash +docker-compose up +``` diff --git a/_examples/simple-app/docker-compose.yml b/_examples/simple-app/docker-compose.yml new file mode 100644 index 000000000..349e67036 --- /dev/null +++ b/_examples/simple-app/docker-compose.yml @@ -0,0 +1,42 @@ +version: '3' +services: + publishing: + image: threedotslabs/golang-librdkafka:1.11.2-stretch + restart: on-failure + depends_on: + - kafka + volumes: + - .:/app + working_dir: /app/publishing-app/ + command: go run main.go + + subscribing: + image: threedotslabs/golang-librdkafka:1.11.2-stretch + restart: on-failure + depends_on: + - kafka + volumes: + - .:/app + working_dir: /app/subscribing-app/ + command: go run main.go + + zookeeper: + image: confluentinc/cp-zookeeper:latest + restart: on-failure + environment: + ZOOKEEPER_CLIENT_PORT: 2181 + logging: + driver: none + + kafka: + image: confluentinc/cp-kafka:latest + restart: on-failure + logging: + driver: none + depends_on: + - zookeeper + environment: + KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true" diff --git a/_examples/simple-app/publishing-app/go.mod b/_examples/simple-app/publishing-app/go.mod new file mode 100644 index 000000000..89c5487dd --- /dev/null +++ b/_examples/simple-app/publishing-app/go.mod @@ -0,0 +1,7 @@ +module main.go + +require ( + github.com/ThreeDotsLabs/watermill v0.1.2 // indirect + github.com/google/uuid v1.1.0 // indirect + github.com/renstrom/shortuuid v3.0.0+incompatible // indirect +) diff --git a/_examples/simple-app/publishing-app/go.sum b/_examples/simple-app/publishing-app/go.sum new file mode 100644 index 000000000..2621d1d9f --- /dev/null +++ b/_examples/simple-app/publishing-app/go.sum @@ -0,0 +1,22 @@ +github.com/ThreeDotsLabs/watermill v0.1.2 h1:1V1SJNRZ7+KYVDX3cFDQGY2jBlMc+tLY0wqhQfCwozQ= +github.com/ThreeDotsLabs/watermill v0.1.2/go.mod h1:c0DOrvvuqbB8uhZlgY/fukFFfv1WZ6HinSktALd9b38= +github.com/confluentinc/confluent-kafka-go v0.11.6 h1:rEblubnNXCjRThwAGnFSzLKYIRAoXLDC3A9r4ciziHU= +github.com/confluentinc/confluent-kafka-go v0.11.6/go.mod h1:u2zNLny2xq+5rWeTQjFHbDzzNuba4P1vo31r9r4uAdg= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-chi/chi v3.3.3+incompatible/go.mod h1:eB3wogJHnLi3x/kFX2A+IbTBlXxmMeXJVKy9tTv1XzQ= +github.com/google/uuid v1.1.0 h1:Jf4mxPC/ziBnoPIdpQdPJ9OeiomAUHLvxmPRSPH9m4s= +github.com/google/uuid v1.1.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= +github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= +github.com/renstrom/shortuuid v3.0.0+incompatible h1:F6T1U7bWlI3FTV+JE8HyeR7bkTeYZJntqQLA9ST4HOQ= +github.com/renstrom/shortuuid v3.0.0+incompatible/go.mod h1:n18Ycpn8DijG+h/lLBQVnGKv1BCtTeXo8KKSbBOrQ8c= +github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= +github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/_examples/simple-app/publishing-app/main.go b/_examples/simple-app/publishing-app/main.go index 72aa5efef..b50e16d72 100644 --- a/_examples/simple-app/publishing-app/main.go +++ b/_examples/simple-app/publishing-app/main.go @@ -2,7 +2,6 @@ package main import ( "encoding/json" - "fmt" "log" "math/rand" "time" @@ -15,34 +14,20 @@ import ( "github.com/renstrom/shortuuid" ) -type postAdded struct { - OccurredOn time.Time `json:"occurred_on"` - - Author string `json:"author"` - Title string `json:"title"` - - Content string `json:"content"` -} - -var letters = []rune("abcdefghijklmnopqrstuvwxyz") - -// randString generates random string of len n -func randString(n int) string { - b := make([]rune, n) - for i := range b { - b[i] = letters[rand.Intn(len(letters))] - } - return string(b) -} +var ( + brokers = []string{"kafka:9092"} +) func main() { - publisher, err := kafka.NewPublisher([]string{"localhost:9092"}, kafka.DefaultMarshaler{}, nil) + log.Println("Starting publishing app") + + publisher, err := kafka.NewPublisher(brokers, kafka.DefaultMarshaler{}, nil) if err != nil { panic(err) } defer publisher.Close() - messagesToAdd := 1000 + messagesToAdd := 10000 workers := 25 msgAdded := make(chan struct{}) @@ -52,8 +37,8 @@ func main() { for range msgAdded { messagesToAdd-- - if messagesToAdd%100000 == 0 { - fmt.Println("left ", messagesToAdd) + if messagesToAdd%1000 == 0 { + log.Println("left ", messagesToAdd) } if messagesToAdd == 0 { allMessagesAdded <- struct{}{} @@ -95,3 +80,23 @@ func main() { // waiting to all being produced <-allMessagesAdded } + +type postAdded struct { + OccurredOn time.Time `json:"occurred_on"` + + Author string `json:"author"` + Title string `json:"title"` + + Content string `json:"content"` +} + +var letters = []rune("abcdefghijklmnopqrstuvwxyz") + +// randString generates random string of len n +func randString(n int) string { + b := make([]rune, n) + for i := range b { + b[i] = letters[rand.Intn(len(letters))] + } + return string(b) +} diff --git a/_examples/simple-app/subscribing-app/counter.go b/_examples/simple-app/subscribing-app/counter.go deleted file mode 100644 index e6f2e3130..000000000 --- a/_examples/simple-app/subscribing-app/counter.go +++ /dev/null @@ -1,52 +0,0 @@ -package main - -import ( - "encoding/json" - "sync/atomic" - - "github.com/ThreeDotsLabs/watermill/message" - "github.com/pkg/errors" - "github.com/satori/go.uuid" -) - -type postsCountUpdated struct { - NewCount int64 `json:"new_count"` -} - -type countStorage interface { - CountAdd() (int64, error) - Count() (int64, error) -} - -type memoryCountStorage struct { - count *int64 -} - -func (m memoryCountStorage) CountAdd() (int64, error) { - return atomic.AddInt64(m.count, 1), nil -} - -func (m memoryCountStorage) Count() (int64, error) { - return atomic.LoadInt64(m.count), nil -} - -type PostsCounter struct { - countStorage countStorage -} - -func (p PostsCounter) Count(msg *message.Message) ([]*message.Message, error) { - // in production use when implementing counter we probably want to make some kind of deduplication here - - newCount, err := p.countStorage.CountAdd() - if err != nil { - return nil, errors.Wrap(err, "cannot add count") - } - - producedMsg := postsCountUpdated{NewCount: newCount} - b, err := json.Marshal(producedMsg) - if err != nil { - return nil, err - } - - return []*message.Message{message.NewMessage(uuid.NewV4().String(), b)}, nil -} diff --git a/_examples/simple-app/subscribing-app/feed_updater.go b/_examples/simple-app/subscribing-app/feed_updater.go deleted file mode 100644 index 91ff5dd95..000000000 --- a/_examples/simple-app/subscribing-app/feed_updater.go +++ /dev/null @@ -1,44 +0,0 @@ -package main - -import ( - "encoding/json" - "fmt" - "time" - - "github.com/ThreeDotsLabs/watermill/message" - "github.com/pkg/errors" -) - -// intentionally not importing type from app1, because we don't need all data and we want to avoid coupling -type postAdded struct { - OccurredOn time.Time `json:"occurred_on"` - Author string `json:"author"` - Title string `json:"title"` -} - -type feedStorage interface { - AddToFeed(title, author string, time time.Time) error -} - -type printFeedStorage struct{} - -func (printFeedStorage) AddToFeed(title, author string, time time.Time) error { - fmt.Printf("Adding to feed: %s by %s @%s\n", title, author, time) - return nil -} - -type FeedGenerator struct { - feedStorage feedStorage -} - -func (f FeedGenerator) UpdateFeed(message *message.Message) ([]*message.Message, error) { - event := postAdded{} - json.Unmarshal(message.Payload, &event) - - err := f.feedStorage.AddToFeed(event.Title, event.Author, event.OccurredOn) - if err != nil { - return nil, errors.Wrap(err, "cannot update feed") - } - - return nil, nil -} diff --git a/_examples/simple-app/subscribing-app/go.mod b/_examples/simple-app/subscribing-app/go.mod new file mode 100644 index 000000000..008adcee9 --- /dev/null +++ b/_examples/simple-app/subscribing-app/go.mod @@ -0,0 +1,7 @@ +module main.go + +require ( + github.com/ThreeDotsLabs/watermill v0.1.2 // indirect + github.com/deathowl/go-metrics-prometheus v0.0.0-20181105123824-7cfe975c505b // indirect + github.com/prometheus/client_golang v0.9.2 // indirect +) diff --git a/_examples/simple-app/subscribing-app/go.sum b/_examples/simple-app/subscribing-app/go.sum new file mode 100644 index 000000000..d5be5dd14 --- /dev/null +++ b/_examples/simple-app/subscribing-app/go.sum @@ -0,0 +1,36 @@ +github.com/ThreeDotsLabs/watermill v0.1.2 h1:1V1SJNRZ7+KYVDX3cFDQGY2jBlMc+tLY0wqhQfCwozQ= +github.com/ThreeDotsLabs/watermill v0.1.2/go.mod h1:c0DOrvvuqbB8uhZlgY/fukFFfv1WZ6HinSktALd9b38= +github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0= +github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= +github.com/confluentinc/confluent-kafka-go v0.11.6 h1:rEblubnNXCjRThwAGnFSzLKYIRAoXLDC3A9r4ciziHU= +github.com/confluentinc/confluent-kafka-go v0.11.6/go.mod h1:u2zNLny2xq+5rWeTQjFHbDzzNuba4P1vo31r9r4uAdg= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/deathowl/go-metrics-prometheus v0.0.0-20181105123824-7cfe975c505b h1:9dj08d9Flaexn3poWqgmV2/LRRhz4rbTMrL6eTGDOAM= +github.com/deathowl/go-metrics-prometheus v0.0.0-20181105123824-7cfe975c505b/go.mod h1:HyiO0WRMVDmaYgeKx/frAiip/fVpUwteTT/RkjwiA0Q= +github.com/go-chi/chi v3.3.3+incompatible/go.mod h1:eB3wogJHnLi3x/kFX2A+IbTBlXxmMeXJVKy9tTv1XzQ= +github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM= +github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU= +github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= +github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= +github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v0.9.2 h1:awm861/B8OKDd2I/6o1dy3ra4BamzKhYOiGItCeZ740= +github.com/prometheus/client_golang v0.9.2/go.mod h1:OsXs2jCmiKlQ1lTBmv21f2mNfw4xf/QclQDMrYNZzcM= +github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910 h1:idejC8f05m9MGOsuEi1ATq9shN03HrxNkD/luQvxCv8= +github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo= +github.com/prometheus/common v0.0.0-20181126121408-4724e9255275 h1:PnBWHBf+6L0jOqq0gIVUe6Yk0/QMZ640k6NvkxcBf+8= +github.com/prometheus/common v0.0.0-20181126121408-4724e9255275/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro= +github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a h1:9a8MnZMP0X2nLJdBg+pBmGgkJlSaKC2KaQmTCk1XDtE= +github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= +github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= +github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/_examples/simple-app/subscribing-app/main.go b/_examples/simple-app/subscribing-app/main.go index 6b8271075..2e2521f29 100644 --- a/_examples/simple-app/subscribing-app/main.go +++ b/_examples/simple-app/subscribing-app/main.go @@ -1,8 +1,21 @@ package main import ( + "encoding/json" + "fmt" + "log" + "net/http" + "os" + "sync/atomic" "time" + "github.com/deathowl/go-metrics-prometheus" + "github.com/pkg/errors" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/rcrowley/go-metrics" + "github.com/satori/go.uuid" + "github.com/ThreeDotsLabs/watermill" "github.com/ThreeDotsLabs/watermill/message" "github.com/ThreeDotsLabs/watermill/message/infrastructure/kafka" @@ -12,7 +25,7 @@ import ( var ( marshaler = kafka.DefaultMarshaler{} - brokers = []string{"localhost:9092"} + brokers = []string{"kafka:9092"} logger = watermill.NewStdLogger(false, false) ) @@ -94,9 +107,9 @@ func main() { func createSubscriber(consumerGroup string, logger watermill.LoggerAdapter) message.Subscriber { sub, err := kafka.NewConfluentSubscriber( kafka.SubscriberConfig{ - Brokers: brokers, - ConsumerGroup: consumerGroup, - ConsumersCount: 8, + Brokers: brokers, + ConsumerGroup: consumerGroup, + ConsumersCount: 8, AutoOffsetReset: "earliest", }, marshaler, @@ -108,3 +121,106 @@ func createSubscriber(consumerGroup string, logger watermill.LoggerAdapter) mess return sub } + +type postsCountUpdated struct { + NewCount int64 `json:"new_count"` +} + +type countStorage interface { + CountAdd() (int64, error) + Count() (int64, error) +} + +type memoryCountStorage struct { + count *int64 +} + +func (m memoryCountStorage) CountAdd() (int64, error) { + return atomic.AddInt64(m.count, 1), nil +} + +func (m memoryCountStorage) Count() (int64, error) { + return atomic.LoadInt64(m.count), nil +} + +type PostsCounter struct { + countStorage countStorage +} + +func (p PostsCounter) Count(msg *message.Message) ([]*message.Message, error) { + // in production use when implementing counter we probably want to make some kind of deduplication here + + newCount, err := p.countStorage.CountAdd() + if err != nil { + return nil, errors.Wrap(err, "cannot add count") + } + + producedMsg := postsCountUpdated{NewCount: newCount} + b, err := json.Marshal(producedMsg) + if err != nil { + return nil, err + } + + return []*message.Message{message.NewMessage(uuid.NewV4().String(), b)}, nil +} + +// intentionally not importing type from app1, because we don't need all data and we want to avoid coupling +type postAdded struct { + OccurredOn time.Time `json:"occurred_on"` + Author string `json:"author"` + Title string `json:"title"` +} + +type feedStorage interface { + AddToFeed(title, author string, time time.Time) error +} + +type printFeedStorage struct{} + +func (printFeedStorage) AddToFeed(title, author string, time time.Time) error { + fmt.Printf("Adding to feed: %s by %s @%s\n", title, author, time) + return nil +} + +type FeedGenerator struct { + feedStorage feedStorage +} + +func (f FeedGenerator) UpdateFeed(message *message.Message) ([]*message.Message, error) { + event := postAdded{} + json.Unmarshal(message.Payload, &event) + + err := f.feedStorage.AddToFeed(event.Title, event.Author, event.OccurredOn) + if err != nil { + return nil, errors.Wrap(err, "cannot update feed") + } + + return nil, nil +} + +func newMetricsMiddleware() middleware.Metrics { + t := metrics.NewTimer() + metrics.Register("handler.time", t) + + errs := metrics.NewCounter() + metrics.Register("handler.errors", errs) + + success := metrics.NewCounter() + metrics.Register("handler.success", success) + + pClient := prometheusmetrics.NewPrometheusProvider( + metrics.DefaultRegistry, + "test", + "subsys", + prometheus.DefaultRegisterer, + 1*time.Second, + ) + go pClient.UpdatePrometheusMetrics() + http.Handle("/metrics", promhttp.Handler()) + + go http.ListenAndServe(":9000", nil) + metricsMiddleware := middleware.NewMetrics(t, errs, success) + metricsMiddleware.ShowStats(time.Second*5, log.New(os.Stderr, "metrics: ", log.Lmicroseconds)) + + return metricsMiddleware +} diff --git a/_examples/simple-app/subscribing-app/metrics.go b/_examples/simple-app/subscribing-app/metrics.go deleted file mode 100644 index 3cd39cd5e..000000000 --- a/_examples/simple-app/subscribing-app/metrics.go +++ /dev/null @@ -1,42 +0,0 @@ -package main - -import ( - "log" - "net/http" - "os" - "time" - - "github.com/ThreeDotsLabs/watermill/message/router/middleware" - "github.com/deathowl/go-metrics-prometheus" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promhttp" - - "github.com/rcrowley/go-metrics" -) - -func newMetricsMiddleware() middleware.Metrics { - t := metrics.NewTimer() - metrics.Register("handler.time", t) - - errs := metrics.NewCounter() - metrics.Register("handler.errors", errs) - - success := metrics.NewCounter() - metrics.Register("handler.success", success) - - pClient := prometheusmetrics.NewPrometheusProvider( - metrics.DefaultRegistry, - "test", - "subsys", - prometheus.DefaultRegisterer, - 1*time.Second, - ) - go pClient.UpdatePrometheusMetrics() - http.Handle("/metrics", promhttp.Handler()) - - go http.ListenAndServe(":9000", nil) - metricsMiddleware := middleware.NewMetrics(t, errs, success) - metricsMiddleware.ShowStats(time.Second*5, log.New(os.Stderr, "metrics: ", log.Lmicroseconds)) - - return metricsMiddleware -} diff --git a/doc.go b/doc.go new file mode 100644 index 000000000..6ac8fe20d --- /dev/null +++ b/doc.go @@ -0,0 +1,13 @@ +// Watermill is a Golang library for working efficiently with message streams. +// +// It is intended for building event driven applications, +// enabling event sourcing, RPC over messages, sagas +// and basically whatever else comes to your mind. +// +// You can use conventional pub/sub implementations +// like Kafka or RabbitMQ, but also HTTP or MySQL binlog if that fits your use case. +// +// Website with detailed documentation: https://watermill.io/ +// +// Getting started guide: https://watermill.io/docs/getting-started/ +package watermill diff --git a/docs/archetypes/default.md b/docs/archetypes/default.md new file mode 100644 index 000000000..00e77bd79 --- /dev/null +++ b/docs/archetypes/default.md @@ -0,0 +1,6 @@ +--- +title: "{{ replace .Name "-" " " | title }}" +date: {{ .Date }} +draft: true +--- + diff --git a/docs/archetypes/docs.md b/docs/archetypes/docs.md new file mode 100644 index 000000000..8e41f2092 --- /dev/null +++ b/docs/archetypes/docs.md @@ -0,0 +1,9 @@ ++++ +title = "" +description = "" +date = {{ .Date }} +weight = 20 +draft = false +bref = "" +toc = true ++++ diff --git a/docs/build.sh b/docs/build.sh new file mode 100755 index 000000000..a4f2e415e --- /dev/null +++ b/docs/build.sh @@ -0,0 +1,38 @@ +set -e -x + +if [ ! -d themes/kube ]; then + mkdir -p themes/kube && pushd themes/kube + git init + git remote add origin https://github.com/jeblister/kube + git fetch --depth 1 origin 5f68bf3e990eff4108fa251f3a3112d081fffba4 + git checkout FETCH_HEAD + popd +fi + +declare -a files_to_link=( + "message/infrastructure/kafka/publisher.go" + "message/infrastructure/kafka/subscriber.go" + "message/infrastructure/kafka/marshaler.go" + "message/infrastructure/gochannel/pubsub.go" + "message/infrastructure/http/subscriber.go" + "message/message.go" + "message/publisher.go" + "message/subscriber.go" + "message/pubsub.go" + "message/router.go" +) + +pushd ../ + +for i in "${files_to_link[@]}" +do + DIR=$(dirname "${i}") + DEST_DIR="docs/content/src-link/${DIR}" + + mkdir -p "${DEST_DIR}" + ln -rsf "./${i}" "./${DEST_DIR}" +done + +popd + +hugo --gc --minify diff --git a/docs/config.toml b/docs/config.toml new file mode 100644 index 000000000..56157fd64 --- /dev/null +++ b/docs/config.toml @@ -0,0 +1,35 @@ +baseURL = "/" +languageCode = "en-us" + +title = "Watermill" +description = "Go library for building event-driven applications." + +theme = "kube" +Paginate = 4 + +pygmentsStyle="autumn" + +googleAnalytics = "UA-128588911-2" + +[Params] + RSSLink = "/index.xml" + author = "Three Dots Labs" + github = "https://github.com/ThreeDotsLabs/" + email = "robert.laszczak@threedotslabs.com" + +[[menu.main]] + name = "Getting Started" + weight = -300 + url = "/docs/getting-started/" +[[menu.main]] + name = "Docs" + weight = -200 + url = "/docs/" +[[menu.main]] + name = "Examples" + weight = 100 + url = "https://github.com/ThreeDotsLabs/watermill/tree/master/_examples" +[[menu.main]] + name = "Support" + weight = 0 + url = "/support/" diff --git a/docs/content/docs/getting-started.md b/docs/content/docs/getting-started.md new file mode 100644 index 000000000..d45a226b6 --- /dev/null +++ b/docs/content/docs/getting-started.md @@ -0,0 +1,238 @@ ++++ +title = "Getting started" +description = "Up and running Watermill" +weight = -9999 +draft = false +toc = true +bref = "Up and running Watermill" +type = "docs" ++++ + +### What is Watermill? + +Watermill is a Golang library for working efficiently with message streams. It is intended for building event-driven applications, enabling event sourcing, RPC over messages, sagas and basically whatever else comes to your mind. You can use conventional pub/sub implementations like Kafka or RabbitMQ, but also HTTP or MySQL binlog if that fits your use case. + +It comes with a set of Pub/Sub, implementations which can be easily replaced by your own implementation + +Watermill is also shipped with the set of standard tools (middlewares) like instrumentation, poison queue, throttling, correlation and other tools used by every message-driven application. + +### Install + +```bash +go get -u github.com/ThreeDotsLabs/watermill/ +``` + +### Subscribing for messages + +One of the most important parts of the Watermill is [*Message*]({{< ref "/docs/message" >}}). It is as important as `http.Request` for `http` package. +Almost every part of Watermill uses this type in some part. + +When we are building reactive/event-driven application/[insert your buzzword here] we always want to listen of incoming messages to react for them. +Watermill is supporting multiple [publishers and subscribers implementations]({{< ref "/docs/pub-sub-implementations" >}}), with compatible interface and abstraction which provide similar behaviour. + +Let's start with subscribing for messages. + +{{% tabs id="subscribing" tabs="go-channel,kafka" labels="Go Channel,Kafka" %}} + +{{% tabs-tab id="go-channel"%}} +{{% load-snippet-partial file="content/docs/getting-started/go-channel/main.go" first_line_contains="import (" last_line_contains="process(messages)" %}} +{{% load-snippet-partial file="content/docs/getting-started/go-channel/main.go" first_line_contains="func process" %}} +{{% /tabs-tab %}} + +{{% tabs-tab id="kafka" %}} +{{% message type="success" %}} +Installed `librdkafka` is required to run Kafka subscriber. +{{% /message %}} + +{{< collapse id="installing_rdkafka" >}} + +{{< collapse-toggle box_id="docker" >}} +Running in Docker +{{% /collapse-toggle %}} +{{% collapse-box id="docker" %}} +Easiest way to run Watermill with Kafka locally is using Docker. + +{{% load-snippet file="content/docs/getting-started/kafka/docker-compose.yml" type="yaml" %}} + +The source should go to `main.go`. + +To run please execute `docker-compose up` command. + +More detailed explanation of how it is running, and how to add live reload you can find in [our [...] article](todo). + +{{% /collapse-box %}} +{{< collapse-toggle box_id="ubuntu" >}} +Installing librdkafka on Ubuntu +{{% /collapse-toggle %}} +{{% collapse-box id="ubuntu" %}} +Newest version of the `librdkafka` for Ubuntu distributions you can find in [Confluent](https://www.confluent.io/)'s repository. + +```bash +# install `software-properties-common`, `wget`, or `gnupg` if not installed yet +sudo apt-get install -y software-properties-common wget gnupg + +# add a new repository +wget -qO - https://packages.confluent.io/deb/4.1/archive.key | sudo apt-key add - +sudo add-apt-repository "deb [arch=amd64] https://packages.confluent.io/deb/4.1 stable main" + +# and then you can install newest version of `librdkafka` +sudo apt-get update && sudo apt-get -y install librdkafka1 librdkafka-dev +``` +{{% /collapse-box %}} +{{< collapse-toggle box_id="redhat" >}} +Installing librdkafka on CentOS/RedHat/Fedora +{{% /collapse-toggle %}} +{{% collapse-box id="redhat" %}} +We will use [Confluent](https://www.confluent.io/)'s repository to download newest version of `librdkafka`. + +```bash +# install `curl` and `which` if not already installed +sudo yum -y install curl which + +# install Confluent public key +sudo rpm --import https://packages.confluent.io/rpm/4.1/archive.key + +# add repository to /etc/yum.repos.d/confluent.repo +sudo cat > /etc/yum.repos.d/confluent.repo << EOF +[Confluent.dist] +name=Confluent repository (dist) +baseurl=https://packages.confluent.io/rpm/4.1/7 +gpgcheck=1 +gpgkey=https://packages.confluent.io/rpm/4.1/archive.key +enabled=1 + +[Confluent] +name=Confluent repository +baseurl=https://packages.confluent.io/rpm/4.1 +gpgcheck=1 +gpgkey=https://packages.confluent.io/rpm/4.1/archive.key +enabled=1 +EOF + +# clean YUM cache +sudo yum clean all + +# install librdkafka +sudo yum -y install librdkafka1 librdkafka-dev +``` + +{{% /collapse-box %}} +{{< collapse-toggle box_id="macOS" >}} +Installing librdkafka on macOS +{{% /collapse-toggle %}} +{{% collapse-box id="macOS" %}} +On macOS, you can install `librdkafka` via [Homebrew](https://brew.sh/): + +```bash +brew install librdkafka +``` +{{% /collapse-box %}} +{{< collapse-toggle box_id="building" >}} +Building from sources (for other distros) +{{% /collapse-toggle %}} +{{% collapse-box id="building" %}} +Manually compiling from sources: + +```bash +wget -O "librdkafka.tar.gz" "https://github.com/edenhill/librdkafka/archive/v0.11.6.tar.gz" + +mkdir -p librdkafka +tar --extract --file "librdkafka.tar.gz" --directory "librdkafka" --strip-components 1 +cd "librdkafka" + +./configure --prefix=/usr && make -j "$(getconf _NPROCESSORS_ONLN)" && make install +``` + +{{% /collapse-box %}} +{{< /collapse >}} + +{{% load-snippet-partial file="content/docs/getting-started/kafka/main.go" first_line_contains="import (" last_line_contains="process(messages)" %}} +{{% load-snippet-partial file="content/docs/getting-started/kafka/main.go" first_line_contains="func process" %}} +{{% /tabs-tab %}} + +{{% /tabs %}} + +### Publishing messages + +{{% tabs id="publishing" tabs="go-channel,kafka" labels="Go Channel,Kafka" %}} + +{{% tabs-tab id="go-channel"%}} +{{% load-snippet-partial file="content/docs/getting-started/go-channel/main.go" first_line_contains="go process(messages)" last_line_contains="publisher.Publish" padding_after="4" %}} +{{% /tabs-tab %}} + +{{% tabs-tab id="kafka" %}} +{{% load-snippet-partial file="content/docs/getting-started/kafka/main.go" first_line_contains="go process(messages)" last_line_contains="publisher.Publish" padding_after="4" %}} +{{% /tabs-tab %}} + +{{% /tabs %}} + +##### Message format + +We don't enforce any message format. You can use strings, JSON, protobuf, Avro, gob or anything else what serializes to `[]byte`. + +### Using *Messages Router* + +[*Publishers and subscribers*]({{< ref "/docs/pub-sub" >}}) are rather low-level parts of Watermill. +In production use, we want usually use something which is higher level and provides some features like [correlation, metrics, poison queue, retrying, throttling etc]({{< ref "/docs/messages-router#middleware" >}}). + +We also don't want to manually send Ack when processing was successful. Sometimes, we also want to send a message after processing another. + +To handle these requirements we created component named [*Router*]({{< ref "/docs/messages-router" >}}). + +The flow of our application looks like this: + +1. We are producing a message to the topic `example.topic_1` every second. +2. `struct_handler` handler is listening to `example.topic_1`. When a message is received, UUID is printed and a new message is produced to `example.topic_2`. +3. `print_events_topic_1` handler is listening to `example.topic_1` and printing message UUID, payload and metadata. Correlation ID should be the same as in message in `example.topic_1`. +4. `print_events_topic_2` handler is listening to `example.topic_2` and printing message UUID, payload and metadata. Correlation ID should be the same as in message in `example.topic_2`. + +#### Router configuration + +For the beginning, we should start with the configuration of the router. We will configure which plugins and middlewares we want to use. + +We also will set up handlers which this router will support. Every handler will independently handle the messages. + +{{% render-md %}} +{{% load-snippet-partial file="content/docs/getting-started/router/main.go" first_line_contains="package" last_line_contains="router.Run()" padding_after="4" %}} +{{% /render-md %}} + +#### Producing messages + +Producing messages work just like before. We only have added `middleware.SetCorrelationID` to set correlation ID. +Correlation ID will be added to all messages produced by the router (`middleware.CorrelationID`). + +{{% render-md %}} +{{% load-snippet-partial file="content/docs/getting-started/router/main.go" first_line_contains="func publishMessages" last_line_contains="time.Sleep(time.Second)" padding_after="2" %}} +{{% /render-md %}} + +#### Handlers + +You may notice that we have two types of *handler functions*: + +1. function `func(msg *message.Message) ([]*message.Message, error)` +2. method `func (c structHandler) Handler(msg *message.Message) ([]*message.Message, error)` + +The second option is useful when our function requires some dependencies like database, logger etc. +When we have just function without dependencies, it's fine to use just a function. + +{{% render-md %}} +{{% load-snippet-partial file="content/docs/getting-started/router/main.go" first_line_contains="func printMessages" last_line_contains="return message.Messages{msg}, nil" padding_after="3" %}} +{{% /render-md %}} + +#### Done! + +We can just run this example by `go run main.go`. + +We just created our first application with Watermill. The full source you can find in [/docs/getting-started/router/main.go](https://github.com/ThreeDotsLabs/watermill/tree/master/content/docs/getting-started/router/main.go). + +### Deployment + +Watermill is not a framework. We don't enforce any type of deployment and it's totally up to you. + +### What's next? + +For more detailed documentation you should check [documentation topics list]({{< ref "/docs" >}}). + +For more advanced examples you should visit [`_examples` directory](https://github.com/ThreeDotsLabs/watermill/tree/master/_examples). + +If anything is not clear feel free to use any of our [support channels]({{< ref "/support" >}}), we will we'll be glad to help. diff --git a/docs/content/docs/getting-started/go-channel/main.go b/docs/content/docs/getting-started/go-channel/main.go new file mode 100644 index 000000000..6c07af28b --- /dev/null +++ b/docs/content/docs/getting-started/go-channel/main.go @@ -0,0 +1,50 @@ +package main + +import ( + "log" + "time" + + "github.com/satori/go.uuid" + + "github.com/ThreeDotsLabs/watermill/message" + + "github.com/ThreeDotsLabs/watermill" + "github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel" +) + +func main() { + pubSub := gochannel.NewGoChannel( + 0, // buffer (channel) size + watermill.NewStdLogger(false, false), + time.Second, // send timeout + ) + + messages, err := pubSub.Subscribe("example.topic") + if err != nil { + panic(err) + } + + go process(messages) + + publishMessages(pubSub) +} + +func publishMessages(publisher message.Publisher) { + for { + msg := message.NewMessage(uuid.NewV4().String(), []byte("Hello, world!")) + + if err := publisher.Publish("example.topic", msg); err != nil { + panic(err) + } + } +} + +func process(messages chan *message.Message) { + for msg := range messages { + log.Printf("received message: %s, payload: %s", msg.UUID, string(msg.Payload)) + + // we need to Acknowledge that we received and processed the message, + // otherwise we will not receive next message + msg.Ack() + } +} diff --git a/docs/content/docs/getting-started/kafka/docker-compose.yml b/docs/content/docs/getting-started/kafka/docker-compose.yml new file mode 100644 index 000000000..f9af05d14 --- /dev/null +++ b/docs/content/docs/getting-started/kafka/docker-compose.yml @@ -0,0 +1,28 @@ +version: '3' +services: + server: + image: threedotslabs/golang-librdkafka:1.11.2-stretch + restart: on-failure + depends_on: + - kafka + volumes: + - .:/app + working_dir: /app + command: go run main.go + + zookeeper: + image: confluentinc/cp-zookeeper:latest + restart: on-failure + environment: + ZOOKEEPER_CLIENT_PORT: 2181 + + kafka: + image: confluentinc/cp-kafka:latest + restart: on-failure + depends_on: + - zookeeper + environment: + KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true" diff --git a/docs/content/docs/getting-started/kafka/go.mod b/docs/content/docs/getting-started/kafka/go.mod new file mode 100644 index 000000000..b3f5421be --- /dev/null +++ b/docs/content/docs/getting-started/kafka/go.mod @@ -0,0 +1,3 @@ +module main.go + +require github.com/ThreeDotsLabs/watermill v0.1.2 // indirect diff --git a/docs/content/docs/getting-started/kafka/go.sum b/docs/content/docs/getting-started/kafka/go.sum new file mode 100644 index 000000000..403242491 --- /dev/null +++ b/docs/content/docs/getting-started/kafka/go.sum @@ -0,0 +1,17 @@ +github.com/ThreeDotsLabs/watermill v0.1.2 h1:1V1SJNRZ7+KYVDX3cFDQGY2jBlMc+tLY0wqhQfCwozQ= +github.com/ThreeDotsLabs/watermill v0.1.2/go.mod h1:c0DOrvvuqbB8uhZlgY/fukFFfv1WZ6HinSktALd9b38= +github.com/confluentinc/confluent-kafka-go v0.11.6 h1:rEblubnNXCjRThwAGnFSzLKYIRAoXLDC3A9r4ciziHU= +github.com/confluentinc/confluent-kafka-go v0.11.6/go.mod h1:u2zNLny2xq+5rWeTQjFHbDzzNuba4P1vo31r9r4uAdg= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-chi/chi v3.3.3+incompatible/go.mod h1:eB3wogJHnLi3x/kFX2A+IbTBlXxmMeXJVKy9tTv1XzQ= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw= +github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= +github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= +github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/docs/content/docs/getting-started/kafka/main.go b/docs/content/docs/getting-started/kafka/main.go new file mode 100644 index 000000000..c6ebb191a --- /dev/null +++ b/docs/content/docs/getting-started/kafka/main.go @@ -0,0 +1,60 @@ +package main + +import ( + "log" + + "github.com/ThreeDotsLabs/watermill" + + "github.com/satori/go.uuid" + + "github.com/ThreeDotsLabs/watermill/message" + "github.com/ThreeDotsLabs/watermill/message/infrastructure/kafka" +) + +func main() { + subscriber, err := kafka.NewConfluentSubscriber( + kafka.SubscriberConfig{ + Brokers: []string{"kafka:9092"}, + ConsumerGroup: "test_consumer_group", + }, + kafka.DefaultMarshaler{}, + watermill.NewStdLogger(false, false), + ) + if err != nil { + panic(err) + } + + messages, err := subscriber.Subscribe("example.topic") + if err != nil { + panic(err) + } + + go process(messages) + + publisher, err := kafka.NewPublisher([]string{"kafka:9092"}, kafka.DefaultMarshaler{}, nil) + if err != nil { + panic(err) + } + + publishMessages(publisher) +} + +func publishMessages(publisher message.Publisher) { + for { + msg := message.NewMessage(uuid.NewV4().String(), []byte("Hello, world!")) + + if err := publisher.Publish("example.topic", msg); err != nil { + panic(err) + } + } +} + +func process(messages chan *message.Message) { + for msg := range messages { + log.Printf("received message: %s, payload: %s", msg.UUID, string(msg.Payload)) + + // we need to Acknowledge that we received and processed the message, + // otherwise we will not receive next message + msg.Ack() + } +} diff --git a/docs/content/docs/getting-started/router/main.go b/docs/content/docs/getting-started/router/main.go new file mode 100644 index 000000000..c200efac2 --- /dev/null +++ b/docs/content/docs/getting-started/router/main.go @@ -0,0 +1,127 @@ +package main + +import ( + "fmt" + "log" + "time" + + "github.com/satori/go.uuid" + + "github.com/ThreeDotsLabs/watermill" + "github.com/ThreeDotsLabs/watermill/message" + "github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel" + "github.com/ThreeDotsLabs/watermill/message/router/middleware" + "github.com/ThreeDotsLabs/watermill/message/router/plugin" +) + +var ( + // just a simplest implementation, + // probably you want to ship your own implementation of `watermill.LoggerAdapter` + logger = watermill.NewStdLogger(false, false) +) + +func main() { + router, err := message.NewRouter(message.RouterConfig{}, logger) + if err != nil { + panic(err) + } + + // this plugin will gracefully shutdown router, when SIGTERM was sent + // you can also close router by just calling `r.Close()` + router.AddPlugin(plugin.SignalsHandler) + + router.AddMiddleware( + // correlation ID will copy correlation id from consumed message metadata to produced messages + middleware.CorrelationID, + + // when error occurred, function will be retried, + // after max retries (or if no Retry middleware is added) Nack is send and message will be resent + middleware.Retry{ + MaxRetries: 3, + WaitTime: time.Millisecond * 100, + Backoff: 3, + Logger: logger, + }.Middleware, + + // this middleware will handle panics from handlers + // and pass them as error to retry middleware in this case + middleware.Recoverer, + ) + + // for simplicity we are using gochannel Pub/Sub here, + // you can replace it with any Pub/Sub implementation, it will work the same + pubSub := gochannel.NewGoChannel(0, logger, time.Second) + + // producing some messages in background + go publishMessages(pubSub) + + if err := router.AddHandler( + "struct_handler", // handler name, must be unique + "example.topic_1", // topic from which we will read events + "example.topic_2", // topic to which we will publish event + pubSub, + structHandler{}.Handler, + ); err != nil { + panic(err) + } + + // just for debug, we are printing all events sent to `example.topic_1` + if err := router.AddNoPublisherHandler( + "print_events_topic_1", + "example.topic_1", + pubSub, + printMessages, + ); err != nil { + panic(err) + } + + // just for debug, we are printing all events sent to `example.topic_2` + if err := router.AddNoPublisherHandler( + "print_events_topic_2", + "example.topic_2", + pubSub, + printMessages, + ); err != nil { + panic(err) + } + + // when everything is ready, let's run router, + // this function is blocking since router is running + if err := router.Run(); err != nil { + panic(err) + } +} + +func publishMessages(publisher message.Publisher) { + for { + msg := message.NewMessage(uuid.NewV4().String(), []byte("Hello, world!")) + middleware.SetCorrelationID(uuid.NewV4().String(), msg) + + log.Printf("sending message %s, correlation id: %s\n", msg.UUID, middleware.MessageCorrelationID(msg)) + + if err := publisher.Publish("example.topic_1", msg); err != nil { + panic(err) + } + + time.Sleep(time.Second) + } +} + +func printMessages(msg *message.Message) ([]*message.Message, error) { + fmt.Printf( + "\n> Received message: %s\n> %s\n> metadata: %v\n\n", + msg.UUID, string(msg.Payload), msg.Metadata, + ) + return nil, nil +} + +type structHandler struct { + // we can add some dependencies here +} + +func (s structHandler) Handler(msg *message.Message) ([]*message.Message, error) { + log.Println("structHandler received message", msg.UUID) + + msg = message.NewMessage(uuid.NewV4().String(), []byte("message produced by structHandler")) + return message.Messages{msg}, nil +} diff --git a/docs/content/docs/message.md b/docs/content/docs/message.md new file mode 100644 index 000000000..4ebbae313 --- /dev/null +++ b/docs/content/docs/message.md @@ -0,0 +1,42 @@ ++++ +title = "Message" +description = "Message is one of core parts of the Watermill" +date = 2018-12-05T12:42:40+01:00 +weight = -1000 +draft = false +bref = "Message is one of core parts of the Watermill" +toc = true ++++ + +### Message + +Message is one of core parts of the Watermill. Messages are emmitted by [*Publishers*]({{< ref "/docs/pub-sub#publisher" >}}) and received by [*Subscribers*]({{< ref "/docs/pub-sub#subscriber" >}}). +When message is processed, you should send [`Ack()`]({{< ref "#ack" >}}) or [`Nack()`]({{< ref "#ack" >}}) when processing failed. + +`Acks` and `Nacks` are processed by Subscribers (in default implementations subscribers are waiting for `Ack` or `Nack`). + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/message.go" first_line_contains="type Message struct {" last_line_contains="ackSentType ackType" padding_after="2" %}} +{{% /render-md %}} + +### Ack + +#### Sending `Ack` + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/message.go" first_line_contains="// Ack" last_line_contains="func (m *Message) Ack() error {" padding_after="0" %}} +{{% /render-md %}} + + +### Nack + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/message.go" first_line_contains="// Nack" last_line_contains="func (m *Message) Nack() error {" padding_after="0" %}} +{{% /render-md %}} + +### Receiving `Ack/Nack` + +{{% render-md %}} +{{% load-snippet-partial file="content/docs/message/receiving-ack.go" first_line_contains="select {" last_line_contains="}" padding_after="0" %}} +{{% /render-md %}} + diff --git a/docs/content/docs/message/receiving-ack.go b/docs/content/docs/message/receiving-ack.go new file mode 100644 index 000000000..d3f714d17 --- /dev/null +++ b/docs/content/docs/message/receiving-ack.go @@ -0,0 +1,24 @@ +package main + +import ( + "log" + "time" + + "github.com/ThreeDotsLabs/watermill/message" +) + +func main() { + msg := message.NewMessage("1", []byte("foo")) + + go func() { + time.Sleep(time.Millisecond * 10) + msg.Ack() + }() + + select { + case <-msg.Acked(): + log.Print("ack received") + case <-msg.Nacked(): + log.Print("nack received") + } +} diff --git a/docs/content/docs/messages-router.md b/docs/content/docs/messages-router.md new file mode 100644 index 000000000..d93ce7037 --- /dev/null +++ b/docs/content/docs/messages-router.md @@ -0,0 +1,106 @@ ++++ +title = "Message Router" +description = "Magic glue of Watermill" +date = 2018-12-05T12:48:04+01:00 +weight = -850 +draft = false +bref = "Magic glue of Watermill" +toc = true ++++ + +[*Publishers and subscribers*]({{< ref "/docs/pub-sub" >}}) are rather low-level parts of Watermill. +In production use, we want usually use something which is higher level and provides some features like [correlation, metrics, poison queue, retrying, throttling, etc.]({{< ref "/docs/messages-router#middleware" >}}). + +We also don't want to send Ack when processing was successful. Sometimes, we also want to send a message after processing another. + +To handle these requirements we created component named Router. + +Kiwi standing on oval + +### Configuration + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="type RouterConfig struct {" last_line_contains="RouterConfig) Validate()" padding_after="2" %}} +{{% /render-md %}} + +### Handler + +At the beginning we need to implement HandlerFunc: + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// HandlerFunc is" last_line_contains="type HandlerFunc func" padding_after="1" %}} +{{% /render-md %}} + +Next we need to add a new handler with `Router.AddHandler`: + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// AddHandler" last_line_contains=") error" padding_after="0" %}} +{{% /render-md %}} + +And example usage from [Getting Started]({{< ref "/docs/getting-started#using-messages-router" >}}): +{{% render-md %}} +{{% load-snippet-partial file="content/docs/getting-started/router/main.go" first_line_contains="if err := router.AddHandler(" last_line_contains="panic(err)" padding_after="1" %}} +{{% /render-md %}} + +### No publisher handler + +Not every handler needs to publish messages. +You can add this kind of handler by using `Router.AddNoPublisherHandler`: + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// AddNoPublisherHandler" last_line_contains=") error" padding_after="0" %}} +{{% /render-md %}} + +### Ack + +You don't need to call `msg.Ack()` or `msg.Nack()` after a message is processed (but you can, of course). +`msg.Ack()` is called when `HanderFunc` doesn't return error. If the error was returned, `msg.Nack()` will be called. + +### Producing messages + +When returning multiple messages in the router, +you should be aware that most of Publisher's implementations don't support [atomically publishing of the messages]({{< ref "/docs/pub-sub#publishing-multiple-messages" >}}). + +It may lead to producing only part of the messages and sending `msg.Nack()` when broker or storage is not available. + +When it is a problem, you should consider publishing maximum one message with one handler. + +### Running router + +To run the router, you need to call `Run()`. + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// Run" last_line_contains="func (r *Router) Run() (err error) {" padding_after="0" %}} +{{% /render-md %}} + +#### Ensuring that router is running + +Sometimes, you want to do something after the router was started. You can use `Running()` method for this. + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// Running" last_line_contains="func (r *Router) Running()" padding_after="0" %}} +{{% /render-md %}} + +### Execution model + +Some *Consumers* may support an only single stream of messages - that means that until `msg.Ack()` is sent you will not receive more messages. + +However, some *Consumers* can, for example, subscribe to multiple partitions in parallel and multiple messages will be sent even previous was not Acked (Kafka Consumer for example). +The router can handle this case and spawn multiple HandlerFunc in parallel. + + +### Middleware + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// HandlerMiddleware" last_line_contains="type HandlerMiddleware" padding_after="1" %}} +{{% /render-md %}} + +A full list of standard middlewares can are in [message/router/middleware](https://github.com/ThreeDotsLabs/watermill/tree/master/message/router/middleware). + +### Plugin + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/router.go" first_line_contains="// RouterPlugin" last_line_contains="type RouterPlugin" padding_after="1" %}} +{{% /render-md %}} + +A full list of standard plugins can are in [message/router/plugin](https://github.com/ThreeDotsLabs/watermill/tree/master/message/router/plugin). diff --git a/docs/content/docs/pub-sub-implementations.md b/docs/content/docs/pub-sub-implementations.md new file mode 100644 index 000000000..bd2524763 --- /dev/null +++ b/docs/content/docs/pub-sub-implementations.md @@ -0,0 +1,175 @@ ++++ +title = "Pub/Sub's implementations" +description = "Golang channel, Kafka, HTTP, Google Cloud Pub/Sub and more!" +date = 2018-12-05T12:47:48+01:00 +weight = -800 +draft = false +bref = "Golang channel, Kafka, HTTP, Google Cloud Pub/Sub and more!" +toc = false ++++ + +| Name | Publisher | Subscriber | Status | +|------|-----------|------------|--------| +| [Golang Channel]({{< ref "#golang-channel" >}}) | x | x | `prod-ready` | +| [Kafka]({{< ref "#kafka" >}}) | x | x | `prod-ready` | +| [HTTP]({{< ref "#http" >}}) | | x | `prod-ready` | +| [Google Cloud Pub/Sub]({{< ref "#google-cloud-pub-sub" >}}) | x | x | [`in-development`](https://github.com/ThreeDotsLabs/watermill/pull/10) | +| MySQL Binlog | | x | [`idea`](https://github.com/ThreeDotsLabs/watermill/issues/5) | + +All built-in implementations can be found in [message/infrastructure](https://github.com/ThreeDotsLabs/watermill/tree/master/message/infrastructure). + +### Golang Channel + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/gochannel/pubsub.go" first_line_contains="// GoChannel" last_line_contains="type GoChannel struct {" %}} +{{% /render-md %}} + +#### Characteristics + +| Feature | Implements | Note | +| ------- | ---------- | ---- | +| ConsumerGroups | no | | +| ExactlyOnceDelivery | yes | | +| GuaranteedOrder | yes | | +| Persistent | no| | + +##### Configuration + +You can inject configuration via the constructor. + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/gochannel/pubsub.go" first_line_contains="func NewGoChannel" last_line_contains="logger:" %}} +{{% /render-md %}} + +#### Publishing + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/gochannel/pubsub.go" first_line_contains="// Publish" last_line_contains="func (g *GoChannel) Publish" %}} +{{% /render-md %}} + +#### Subscribing + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/gochannel/pubsub.go" first_line_contains="// Subscribe" last_line_contains="func (g *GoChannel) Subscribe" %}} +{{% /render-md %}} + +#### Marshaler + +No marshaling is needed when sending messages within the process. + +### Kafka + +Kafka is one of the most popular Pub/Subs. We are providing Pub/Sub implementation based on [Confluent's bindings to `librdkafka`](https://github.com/confluentinc/confluent-kafka-go). + +`librdkafka` is required to run Kafka Pub/Sub. Installation guide can be found in [Getting Started]({{< ref "/docs/getting-started#subscribing_kafka" >}}). + +#### Characteristics + +| Feature | Implements | Note | +| ------- | ---------- | ---- | +| ConsumerGroups | yes | | +| ExactlyOnceDelivery | no | in theory can be achieved with [Transactions](https://www.confluent.io/blog/transactions-apache-kafka/), currently no support for any Golang client | +| GuaranteedOrder | yes | require [paritition key usage](#using-partition-key) | +| Persistent | yes| | + +#### Configuration + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/subscriber.go" first_line_contains="type SubscriberConfig struct" last_line_contains="func (c SubscriberConfig)" %}} +{{% /render-md %}} + +##### Passing custom `librdkafka` config + +You can pass custom config parameters (for example SSL Configuration) via `KafkaConfigOverwrite` in `SubscriberConfig` and `kafkaConfigOverwrite` to `NewPublisher`. + +You can find a list of available options in [librdkafka documentation](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md). + +#### Publishing + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/publisher.go" first_line_contains="// Publish" last_line_contains="func (p confluentPublisher) Publish" %}} +{{% /render-md %}} + +#### Subscribing + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/subscriber.go" first_line_contains="// Subscribe" last_line_contains="func (s *confluentSubscriber) Subscribe" %}} +{{% /render-md %}} + +#### Marshaler + +Watermill's messages cannot be directly sent to Kafka - they need to be marshaled. You can implement your marshaler or use default implementation. + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/marshaler.go" first_line_contains="// Marshaler" last_line_contains="func (DefaultMarshaler)" padding_after="0" %}} +{{% /render-md %}} + +#### Partitioning + +Our Publisher has support for the partitioning mechanism. + +It can be done with special Marshaler implementation: + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/kafka/marshaler.go" first_line_contains="type kafkaJsonWithPartitioning" last_line_contains="func (j kafkaJsonWithPartitioning) Marshal" padding_after="0" %}} +{{% /render-md %}} + +When using, you need to pass your function to generate partition key. +It's a good idea to pass this partition key with metadata to not unmarshal entire message. + +{{< highlight >}} +marshaler := kafka.NewWithPartitioningMarshaler(func(topic string, msg *message.Message) (string, error) { + return msg.Metadata.Get("partition"), nil +}) +{{< /highlight >}} + +### HTTP + +For this moment only HTTP subscriber is available. There is an issue for a [HTTP publisher](https://github.com/ThreeDotsLabs/watermill/issues/17). + +HTTP subscriber allows us to send messages received by HTTP request (for example - webhooks). +You can then post them to any Publisher. Here is an example with [sending HTTP messages to Kafka](https://github.com/ThreeDotsLabs/watermill/blob/master/_examples/http-to-kafka/main.go). + +When implemented, HTTP publisher can be used as webhooks sender. + +#### Characteristics + +| Feature | Implements | Note | +| ------- | ---------- | ---- | +| ConsumerGroups | no | | +| ExactlyOnceDelivery | no | | +| GuaranteedOrder | yes | | +| Persistent | no| | + +#### Configuration + +The configuration of HTTP subscriber is done via the constructor. + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/http/subscriber.go" first_line_contains="// NewSubscriber" last_line_contains="func NewSubscriber(" %}} +{{% /render-md %}} + +#### Running + +To run HTTP subscriber you need to run `StartHTTPServer()`. It needs to be run after `Subscribe()`. + +When using with the router, you should wait for the router to start. + +{{< highlight >}} +<-r.Running() +httpSubscriber.StartHTTPServer() +{{< /highlight >}} + +#### Subscribing + +{{% render-md %}} +{{% load-snippet-partial file="content/src-link/message/infrastructure/http/subscriber.go" first_line_contains="// Subscribe adds" last_line_contains="func (s *Subscriber) Subscribe" %}} +{{% /render-md %}} + +### Google Cloud Pub/Sub + +In progress. + +### Implementing your own Pub/Sub + +There aren't your Pub/Sub implementation? Please check [Implementing custom Pub/Sub]({{< ref "pub-sub-implementing" >}}). diff --git a/docs/content/docs/pub-sub-implementing.md b/docs/content/docs/pub-sub-implementing.md new file mode 100644 index 000000000..29dc15595 --- /dev/null +++ b/docs/content/docs/pub-sub-implementing.md @@ -0,0 +1,39 @@ ++++ +title = "Implementing custom Pub/Sub" +description = "Bring Your Own Pub/Sub" +date = 2018-12-05T12:48:34+01:00 +weight = -300 +draft = false +bref = "Bring Your Own Pub/Sub" +toc = true ++++ + +### Pub/Sub interface + +In simple words - to implement Pub/Sub you need to implement `message.PubSub` interface. + +{{% render-md %}} +{{% load-snippet file="content/src-link/message/publisher.go" %}} +{{% load-snippet file="content/src-link/message/subscriber.go" %}} +{{% load-snippet-partial file="content/src-link/message/pubsub.go" first_line_contains="type PubSub interface {" last_line_contains="Close() error" padding_after="1" %}} +{{% /render-md %}} + +### TODO list + +But they are some things about which you cannot forget: + +1. Good logging +2. Replaceable and configureable messages marshaler +3. `Close()` implementation for publisher and subscriber which is: + - idempotent + - working event when publisher or subscriber is blocked (for example: when waiting for Ack) + - working when subscriber output channel is blocked (because nothing is listening for it) +4. `Ack()` **and** `Nack()` support for consumed messages +5. Redelivery on `Nack()` on consumed message +6. Use [Universal Pub/Sub tests]({{< ref "/docs/pub-sub#universal-tests" >}}) +7. Performance optimizations +8. godoc's, [Markdown docs]({{< ref "/docs/pub-sub-implementations" >}}) and [examples Getting Started](/docs/getting-started) + +We will be also thankful for submitting [merge requests](https://github.com/ThreeDotsLabs/watermill/pulls) with new Pub/Subs implementation. + +If anything is not clear feel free to use any of our [support channels]({{< ref "/support" >}}), we will we'll be glad to help. diff --git a/docs/content/docs/pub-sub.md b/docs/content/docs/pub-sub.md new file mode 100644 index 000000000..f8492e416 --- /dev/null +++ b/docs/content/docs/pub-sub.md @@ -0,0 +1,80 @@ ++++ +title = "Pub/Sub" +description = "Publishers and Subscribers" +date = 2018-12-05T12:47:30+01:00 +weight = -900 +draft = false +bref = "Publishers and Subscribers" +toc = true ++++ + +### Publisher + +{{% render-md %}} +{{% load-snippet file="content/src-link/message/publisher.go" %}} +{{% /render-md %}} + +#### Publishing multiple messages + +Most publishers implementations don't support atomic publishing of messages. +That means, that when publishing one of the messages failed the next messages will be not published. + +#### Async publish + +Publish can be synchronous or asynchronous - it depends on implementation. + +#### `Close()` + +Close should flush unsent messages if the publisher is async. +**It's important to not forget to close subscriber**, in the other hand you may lose some of the messages. + +### Subscriber + +{{% render-md %}} +{{% load-snippet file="content/src-link/message/subscriber.go" %}} +{{% /render-md %}} + +#### Ack/Nack mechanism + +It is *Subscriber's* responsibility to handle `Ack` and `Nack` from a message. +A proper implementation should wait for `Ack` or `Nack` before consuming the next message. + +**Important Subscriber's implementation notice**: +Ack/offset to message's storage/broker **must** be sent after Ack from Watermill's message. +If it wouldn't, it is a possibility to lose messages when the process will die before messages were processed. + +#### `Close()` + +Close closes all subscriptions with their output channels and flush offsets etc. when needed. + +### At-least-once delivery + +Watermill is build with [at-least-once delivery](http://www.cloudcomputingpatterns.org/at_least_once_delivery/) semantics. +That means, that when some error with occur when processing message and Ack cannot be sent the message will be redelivered. + +You need to keep it in mind and build your application to be [idempotent](http://www.cloudcomputingpatterns.org/idempotent_processor/) or implement deduplication mechanism. + +Unfortunately, it's not possible to create universal [*middleware*]({{< ref "/docs/messages-router#middleware" >}}) for deduplication but we encourage you to make your own. + +### Universal tests + +Every Pub/Sub is similar. +To don't implement separated tests for every Pub/Sub we create test suite which should be passed by any Pub/Sub implementation. + +These tests can be found in `message/infrastructure/test_pubsub.go`. + +### Built-in implementations + +To check available Pub/Subs implementation please check [Pub/Sub's implementations]({{< ref "/docs/pub-sub-implementations" >}}) + +### Implementing custom Pub/Sub + +When any implementation of Pub/Sub. [Implementing custom Pub/Sub]({{< ref "/docs/pub-sub-implementing" >}}). + +We will be also thankful for submitting [merge requests](https://github.com/ThreeDotsLabs/watermill/pulls) with new Pub/Subs implementation. + +You can also request new Pub/Sub implementation by submitting a [new issue](https://github.com/ThreeDotsLabs/watermill/issues). + +### Keep going! + +When you already know, how Pub/Sub is working we recommend to check [*Message Router component*]({{< ref "/docs/messages-router" >}}). diff --git a/docs/content/docs/troubleshooting.md b/docs/content/docs/troubleshooting.md new file mode 100644 index 000000000..f97aaa09e --- /dev/null +++ b/docs/content/docs/troubleshooting.md @@ -0,0 +1,66 @@ ++++ +title = "Troubleshooting" +description = "When something went wrong" +weight = 0 +draft = false +toc = true +bref = "When something went wrong" +type = "docs" ++++ + +### Logging + +In most cases, you will find the answer to your problems in a log. +Watermill offers a significant amount of logs of different severity levels. + +If you are using `StdLoggerAdapter`, just change `debug`, and `trace` options to true: + +{{< highlight >}} +logger := watermill.NewStdLogger(true, true) +{{< /highlight >}} + +### I had a deadlock + +When running locally, you can run: + +- `CTRL + \` for Linux +- `kill -s SIGQUIT [pid]` for another UNIX systems + + +to send `SIGQUIT` to the process. +It will kill the process and print all goroutines with the line on which they are now. + +``` +SIGQUIT: quit +PC=0x45e7c3 m=0 sigcode=128 + +goroutine 1 [runnable]: +github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel.(*GoChannel).sendMessage(0xc000024100, 0x7c5250, 0xd, 0xc000872d70, 0x0, 0x0) + /home/example/go/src/github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel/pubsub.go:83 +0x36a +github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel.(*GoChannel).Publish(0xc000024100, 0x7c5250, 0xd, 0xc000098530, 0x1, 0x1, 0x0, 0x0) + /home/example/go/src/github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel/pubsub.go:53 +0x6d +main.publishMessages(0x7fdf7a317000, 0xc000024100) + /home/example/go/src/github.com/ThreeDotsLabs/watermill/docs/content/docs/getting-started/go-channel/main.go:43 +0x1ec +main.main() + /home/example/go/src/github.com/ThreeDotsLabs/watermill/docs/content/docs/getting-started/go-channel/main.go:36 +0x20a + +// ... +``` + +When running in production, and you don't want to kill the entire process it is a good idea to use [pprof](https://golang.org/pkg/net/http/pprof/). + +You can visit [http://localhost:6060/debug/pprof/goroutine?debug=1](http://localhost:6060/debug/pprof/goroutine?debug=1) to find all goroutines status. + + +``` +goroutine profile: total 5 +1 @ 0x41024c 0x6a8311 0x6a9bcb 0x6a948d 0x7028bc 0x70260a 0x42f187 0x45c971 +# 0x6a8310 github.com/ThreeDotsLabs/watermill.LogFields.Add+0xd0 /home/example/go/src/github.com/ThreeDotsLabs/watermill/log.go:15 +# 0x6a9bca github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel.(*GoChannel).sendMessage+0x6fa /home/example/go/src/github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel/pubsub.go:75 +# 0x6a948c github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel.(*GoChannel).Publish+0x6c /home/example/go/src/github.com/ThreeDotsLabs/watermill/message/infrastructure/gochannel/pubsub.go:53 +# 0x7028bb main.publishMessages+0x1eb /home/example/go/src/github.com/ThreeDotsLabs/watermill/docs/content/docs/getting-started/go-channel/main.go:43 +# 0x702609 main.main+0x209 /home/example/go/src/github.com/ThreeDotsLabs/watermill/docs/content/docs/getting-started/go-channel/main.go:36 +# 0x42f186 runtime.main+0x206 /usr/lib/go/src/runtime/proc.go:201 + +// ... +``` \ No newline at end of file diff --git a/docs/content/support.md b/docs/content/support.md new file mode 100644 index 000000000..5039a5f81 --- /dev/null +++ b/docs/content/support.md @@ -0,0 +1,14 @@ ++++ +title = "Support" +description = "" ++++ + +### Community Support + +Please join us on the `#watermill` channel on the [Gophers slack](https://gophers.slack.com/): You can get the invite [here](https://gophersinvite.herokuapp.com/). + +### Professional Support + +For enterprise support please contact us by e-mail: **{{% contact-email %}}**. + +You can also use the [contact form](https://threedotslabs.com/#contact-form) on our website. diff --git a/docs/layouts/index.html b/docs/layouts/index.html new file mode 100644 index 000000000..448fdbfd9 --- /dev/null +++ b/docs/layouts/index.html @@ -0,0 +1,65 @@ +{{ define "title"}} {{ .Site.Title}} {{end}} +{{ define "header"}} {{ partial "header" .}} {{end}} +{{ define "main"}} + + + +
+
+

{{.Title}}

+

Go library for building event-driven applications.

+
+ +
+
+
+
+ +
+

Easy to use

+

Our goal was to create a tool which is easy to understand, even not by senior developers.

+
+
+
+ +
+

Universal

+

It doesn't matter that you want to do Event-driven architecture, CQRS, Event Sourcing or just stream MySQL Binlog to Kafka.

+
+
+
+ +
+

Fast

+

Watermill was designed to process hundreds of thousands of messages per second.

+
+
+
+
+
+ +
+

Flexible

+

Every component is built in a way, that allows you to configure it for your needs. You can also implement your middlewares to the router.

+
+
+
+ +
+

Resilient

+

Watermill is using proven technologies and has a strong unit and integration tests coverage for the critical functionalities.

+
+
+ +
+
+ A detailed explanation of why we build Watermill you can find in our Introducing Watermill blog post. +
+
+
+ +{{ end }} +{{ define "footer"}} {{ partial "footer" .}} {{end}} diff --git a/docs/layouts/partials/footer.html b/docs/layouts/partials/footer.html new file mode 100644 index 000000000..80ae1b269 --- /dev/null +++ b/docs/layouts/partials/footer.html @@ -0,0 +1,14 @@ + \ No newline at end of file diff --git a/docs/layouts/shortcodes/collapse-box.html b/docs/layouts/shortcodes/collapse-box.html new file mode 100644 index 000000000..ed7e6898e --- /dev/null +++ b/docs/layouts/shortcodes/collapse-box.html @@ -0,0 +1,3 @@ +
+ {{ .Inner }} +
\ No newline at end of file diff --git a/docs/layouts/shortcodes/collapse-toggle.html b/docs/layouts/shortcodes/collapse-toggle.html new file mode 100644 index 000000000..d541b5490 --- /dev/null +++ b/docs/layouts/shortcodes/collapse-toggle.html @@ -0,0 +1 @@ +

{{ .Inner }}

\ No newline at end of file diff --git a/docs/layouts/shortcodes/collapse.html b/docs/layouts/shortcodes/collapse.html new file mode 100644 index 000000000..95d838c70 --- /dev/null +++ b/docs/layouts/shortcodes/collapse.html @@ -0,0 +1,3 @@ +
+ {{ .Inner }} +
\ No newline at end of file diff --git a/docs/layouts/shortcodes/contact-email.html b/docs/layouts/shortcodes/contact-email.html new file mode 100644 index 000000000..7e80368d6 --- /dev/null +++ b/docs/layouts/shortcodes/contact-email.html @@ -0,0 +1,3 @@ +{{/* hack: we cannot use variables directly on the markdown, only shortcodes */}} + +{{ $.Page.Site.Params.Email }} \ No newline at end of file diff --git a/docs/layouts/shortcodes/highlight.html b/docs/layouts/shortcodes/highlight.html new file mode 100644 index 000000000..8df83db38 --- /dev/null +++ b/docs/layouts/shortcodes/highlight.html @@ -0,0 +1 @@ +{{ highlight .Inner (.Get "language" | default "go") "" }} \ No newline at end of file diff --git a/docs/layouts/shortcodes/load-snippet-partial.html b/docs/layouts/shortcodes/load-snippet-partial.html new file mode 100644 index 000000000..8fa952c88 --- /dev/null +++ b/docs/layouts/shortcodes/load-snippet-partial.html @@ -0,0 +1,67 @@ +{{ $file := (.Get "file") }} +{{ $content := readFile $file }} + +{{ $first_line_contains := (.Get "first_line_contains") }} +{{ $last_line_contains := (.Get "last_line_contains") }} + +{{ $show_line := false }} + +{{/*if true, first or last line was not found*/}} +{{ $first_line_found := false}} +{{ $last_line_found := false}} + +{{ $padding_after := (.Get "padding_after" | default "0" | int) }} + +{{ $first_line_num := 0 }} +{{ $last_line_num := 0 }} + +{{ $linkFile := $file }} +{{ if in $file "content/src-link/" }} + {{ $linkFile = replace $linkFile "content/src-link/" "" }} +{{ end }} + +{{ $lines := slice }} + +{{ range $elem_key, $elem_val := split $content "\n" }} + {{ $line_num := (add $elem_key 1) }} + + {{ if and (not $first_line_found) (in $elem_val $first_line_contains) }} + {{ if ne $elem_key 0 }} + {{ $lines = $lines | append "// ..." }} + {{ end }} + + {{ $show_line = true }} + {{ $first_line_found = true}} + {{ $first_line_num = $line_num }} + {{ end }} + + {{ if $show_line }} + {{ $lines = $lines | append $elem_val }} + {{ end }} + + {{ if and ($first_line_found) (in $elem_val $last_line_contains) (ne $last_line_contains "") }} + {{ $last_line_found = true }} + {{ end }} + + {{ if and $last_line_found $show_line }} + {{ if gt $padding_after 0 }} + {{ $padding_after = sub $padding_after 1}} + {{ else }} + {{ $lines = $lines | append "// ..." }} + {{ $show_line = false }} + {{ $last_line_num = $line_num }} + {{ end }} + {{ end }} +{{ end }} + +Full source: [{{ $linkFile }}](https://github.com/ThreeDotsLabs/watermill/tree/master/{{ $linkFile }}{{ if ne $first_line_num 0 }}#L{{ $first_line_num }}{{ end }}) + +{{ highlight (delimit $lines "\n") (.Get "type" | default "go") "" }} + +{{if not $first_line_found }} + {{ errorf "`first_line_contains` %s not found in %s snippet" $first_line_contains $file }} +{{end}} + +{{if and (not $last_line_found) (ne $last_line_contains "") }} + {{ errorf "`last_line_contains` %s not found in %s snippet" $last_line_contains $file }} +{{end}} \ No newline at end of file diff --git a/docs/layouts/shortcodes/load-snippet.html b/docs/layouts/shortcodes/load-snippet.html new file mode 100644 index 000000000..032f980c6 --- /dev/null +++ b/docs/layouts/shortcodes/load-snippet.html @@ -0,0 +1,26 @@ +{{ $file := (.Get "file") }} +{{ $content := readFile $file }} + +{{ $start_line := (.Get "start_line") | default "0" }} +{{ $end_line := (.Get "end_line") | default "0" }} + +{{ $has_start_line := (ne $start_line "0") }} +{{ $has_end_line := (ne $end_line "0") }} + +{{ $lines := slice }} + +{{ $linkFile := $file }} +{{ if in $file "content/src-link/" }} + {{ $linkFile = replace $linkFile "content/src-link/" "" }} +{{ end }} + +Full source: [{{ $linkFile }}](https://github.com/ThreeDotsLabs/watermill/tree/master/{{ $linkFile }}) + +{{ range $elem_key, $elem_val := split $content "\n" }} + {{if and (or (not $has_start_line) (ge (add $elem_key 1) ($start_line | int))) (or (not $has_end_line) (le (add $elem_key 1) ($end_line | int)))}} + {{ $lines = $lines | append $elem_val }} + {{ end }} +{{ end }} + + +{{ highlight (delimit $lines "\n") (.Get "type" | default "go") "" }} diff --git a/docs/layouts/shortcodes/message.html b/docs/layouts/shortcodes/message.html new file mode 100644 index 000000000..16fc58452 --- /dev/null +++ b/docs/layouts/shortcodes/message.html @@ -0,0 +1,4 @@ +
+ {{ if ne (.Get "title") "" }}
{{ .Get "title" }}
{{ end }} + {{ .Inner }} +
\ No newline at end of file diff --git a/docs/layouts/shortcodes/render-md.html b/docs/layouts/shortcodes/render-md.html new file mode 100644 index 000000000..825c3f939 --- /dev/null +++ b/docs/layouts/shortcodes/render-md.html @@ -0,0 +1,11 @@ +{{/* +hugo pls, why i just cannot render shortcode as markdown, +used for: + +{{% render-md %}} +{{% load-snippet-partial (...) %}} +{{% /render-md %}} + +bacause it is rendered as raw text by default +*/}} +{{.Inner}} \ No newline at end of file diff --git a/docs/layouts/shortcodes/tabs-tab.html b/docs/layouts/shortcodes/tabs-tab.html new file mode 100644 index 000000000..00885f920 --- /dev/null +++ b/docs/layouts/shortcodes/tabs-tab.html @@ -0,0 +1,3 @@ +
+ {{ .Inner }} +
diff --git a/docs/layouts/shortcodes/tabs.html b/docs/layouts/shortcodes/tabs.html new file mode 100644 index 000000000..657af6448 --- /dev/null +++ b/docs/layouts/shortcodes/tabs.html @@ -0,0 +1,13 @@ +
+ + {{.Inner}} +
\ No newline at end of file diff --git a/docs/static/css/custom.css b/docs/static/css/custom.css new file mode 100644 index 000000000..b9dd3a09c --- /dev/null +++ b/docs/static/css/custom.css @@ -0,0 +1,24 @@ +pre, pre code { + background: #f8f8f8 !important; +} + +#kube-features .row:first-child { + border-bottom: none; +} + +.message > p { + margin-bottom: 0px; +} + +/* fix for stupid css */ +.my-collapse div { + border: none; + padding: 0; + margin-bottom: 0; +} + +.my-collapse > div.collapse-box { + border: 1px solid rgba(0, 0, 0, 0.1); + padding: 24px 32px 1px; + margin-bottom: 1px; +} \ No newline at end of file diff --git a/docs/static/img/watermill-router.svg b/docs/static/img/watermill-router.svg new file mode 100644 index 000000000..5b174c939 --- /dev/null +++ b/docs/static/img/watermill-router.svg @@ -0,0 +1 @@ +PublisherQueue / Broker / TopicSubscriberRouterMiddlewaresHandlerPublisherQueue / Broker / TopicWatermillAck or Nack \ No newline at end of file diff --git a/log.go b/log.go index 2630d9919..74ed4db2e 100644 --- a/log.go +++ b/log.go @@ -45,7 +45,7 @@ type StdLoggerAdapter struct { func NewStdLogger(debug, trace bool) LoggerAdapter { l := log.New(os.Stderr, "[watermill] ", log.LstdFlags|log.Lmicroseconds|log.Lshortfile) - a := &StdLoggerAdapter{InfoLogger: l} + a := &StdLoggerAdapter{InfoLogger: l, ErrorLogger: l} if debug { a.DebugLogger = l @@ -58,7 +58,7 @@ func NewStdLogger(debug, trace bool) LoggerAdapter { } func (l *StdLoggerAdapter) Error(msg string, err error, fields LogFields) { - l.log(l.TraceLogger, "ERROR", msg, fields.Add(LogFields{"err": err})) + l.log(l.ErrorLogger, "ERROR", msg, fields.Add(LogFields{"err": err})) } func (l *StdLoggerAdapter) Info(msg string, fields LogFields) { diff --git a/message/infrastructure/gochannel/pubsub.go b/message/infrastructure/gochannel/pubsub.go index 8b778dbfb..14ab2d7a8 100644 --- a/message/infrastructure/gochannel/pubsub.go +++ b/message/infrastructure/gochannel/pubsub.go @@ -16,7 +16,12 @@ type subscriber struct { outputChannel chan *message.Message } -type goChannel struct { +// GoChannel is the simplest Pub/Sub implementation. +// It is based on Golang's channels which are sent within the process. +// +// GoChannel has no global state, +// that means that you need to use the same instance for Publishing and Subscribing! +type GoChannel struct { sendTimeout time.Duration buffer int64 @@ -29,7 +34,7 @@ type goChannel struct { } func NewGoChannel(buffer int64, logger watermill.LoggerAdapter, sendTimeout time.Duration) message.PubSub { - return &goChannel{ + return &GoChannel{ sendTimeout: sendTimeout, buffer: buffer, @@ -39,7 +44,11 @@ func NewGoChannel(buffer int64, logger watermill.LoggerAdapter, sendTimeout time } } -func (g *goChannel) Publish(topic string, messages ...*message.Message) error { +// Publish in GoChannel is blocking until all consumers consume and acknowledge the message. +// Sending message to one subscriber has timeout equal to GoChannel.sendTimeout configured via constructor. +// +// Messages are not persisted. If there are no subscribers and message is produced it will be gone. +func (g *GoChannel) Publish(topic string, messages ...*message.Message) error { for _, msg := range messages { if err := g.sendMessage(topic, msg); err != nil { return err @@ -49,7 +58,7 @@ func (g *goChannel) Publish(topic string, messages ...*message.Message) error { return nil } -func (g *goChannel) sendMessage(topic string, message *message.Message) error { +func (g *GoChannel) sendMessage(topic string, message *message.Message) error { messageLogFields := watermill.LogFields{ "message_uuid": message.UUID, } @@ -92,7 +101,11 @@ func (g *goChannel) sendMessage(topic string, message *message.Message) error { return nil } -func (g *goChannel) Subscribe(topic string) (chan *message.Message, error) { +// Subscribe returns channel to which all published messages are sent. +// Messages are not persisted. If there are no subscribers and message is produced it will be gone. +// +// There are no consumer groups support etc. Every consumer will receive every produced message. +func (g *GoChannel) Subscribe(topic string) (chan *message.Message, error) { g.subscribersLock.Lock() defer g.subscribersLock.Unlock() @@ -109,7 +122,7 @@ func (g *goChannel) Subscribe(topic string) (chan *message.Message, error) { return s.outputChannel, nil } -func (g *goChannel) Close() error { +func (g *GoChannel) Close() error { g.subscribersLock.Lock() defer g.subscribersLock.Unlock() diff --git a/message/infrastructure/http/subscriber.go b/message/infrastructure/http/subscriber.go index fe74ff5fa..10128a6a0 100644 --- a/message/infrastructure/http/subscriber.go +++ b/message/infrastructure/http/subscriber.go @@ -11,6 +11,7 @@ import ( type UnmarshalMessageFunc func(topic string, request *http.Request) (*message.Message, error) +// Subscriber can subscribe to HTTP requests and create Watermill's messages based on them. type Subscriber struct { router chi.Router server *http.Server @@ -24,6 +25,13 @@ type Subscriber struct { closed bool } +// NewSubscriber creates new Subscriber. +// +// addr is TCP address to listen on +// +// unmarshalMessageFunc is function which converts HTTP request to Watermill's message. +// +// logger is Watermill's logger. func NewSubscriber(addr string, unmarshalMessageFunc UnmarshalMessageFunc, logger watermill.LoggerAdapter) (*Subscriber, error) { r := chi.NewRouter() s := &http.Server{Addr: addr, Handler: r} @@ -39,17 +47,23 @@ func NewSubscriber(addr string, unmarshalMessageFunc UnmarshalMessageFunc, logge }, nil } -func (s *Subscriber) Subscribe(topic string) (chan *message.Message, error) { +// Subscribe adds HTTP handler which will listen in provided url for messages. +// +// Subscribe needs to be called before `StartHTTPServer`. +// +// When request is sent, it will wait for the `Ack`. When Ack is received 200 HTTP status wil be sent. +// When Nack is sent, 500 HTTP status will be sent. +func (s *Subscriber) Subscribe(url string) (chan *message.Message, error) { messages := make(chan *message.Message) s.outputChannelsLock.Lock() s.outputChannels = append(s.outputChannels, messages) s.outputChannelsLock.Unlock() - baseLogFields := watermill.LogFields{"topic": topic} + baseLogFields := watermill.LogFields{"url": url} - s.router.Post(topic, func(w http.ResponseWriter, r *http.Request) { - msg, err := s.unmarshalMessageFunc(topic, r) + s.router.Post(url, func(w http.ResponseWriter, r *http.Request) { + msg, err := s.unmarshalMessageFunc(url, r) if err != nil { s.logger.Info("Cannot unmarshal message", baseLogFields.Add(watermill.LogFields{"err": err})) w.WriteHeader(http.StatusBadRequest) diff --git a/message/infrastructure/kafka/marshaler.go b/message/infrastructure/kafka/marshaler.go index 8b87c598b..9eb08ec7a 100644 --- a/message/infrastructure/kafka/marshaler.go +++ b/message/infrastructure/kafka/marshaler.go @@ -6,10 +6,14 @@ import ( "github.com/pkg/errors" ) +const UUIDHeaderKey = "_watermill_message_uuid" + +// Marshaler marshals Watermill's message to Kafka message. type Marshaler interface { Marshal(topic string, msg *message.Message) (*confluentKafka.Message, error) } +// Unmarshaler unmarshals Kafka's message to Watermill's message. type Unmarshaler interface { Unmarshal(*confluentKafka.Message) (*message.Message, error) } @@ -19,8 +23,6 @@ type MarshalerUnmarshaler interface { Unmarshaler } -const UUIDHeaderKey = "_watermill_message_uuid" - type DefaultMarshaler struct{} func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*confluentKafka.Message, error) { diff --git a/message/infrastructure/kafka/publisher.go b/message/infrastructure/kafka/publisher.go index 47af1bc62..5bea9554f 100644 --- a/message/infrastructure/kafka/publisher.go +++ b/message/infrastructure/kafka/publisher.go @@ -21,7 +21,7 @@ func NewPublisher(brokers []string, marshaler Marshaler, kafkaConfigOverwrite ka "bootstrap.servers": strings.Join(brokers, ","), "queue.buffering.max.messages": 10000000, "queue.buffering.max.kbytes": 2097151, - "debug": ",", + "debug": ",", } if err := mergeConfluentConfigs(config, kafkaConfigOverwrite); err != nil { @@ -40,6 +40,10 @@ func NewCustomPublisher(producer *kafka.Producer, marshaler Marshaler) (message. return &confluentPublisher{producer, marshaler, false}, nil } +// Publish publishes message to Kafka. +// +// Publish is blocking and wait for ack from Kafka. +// When one of messages delivery fails - function is interrupted. func (p confluentPublisher) Publish(topic string, msgs ...*message.Message) error { if p.closed { return errors.New("publisher closed") diff --git a/message/infrastructure/kafka/subscriber.go b/message/infrastructure/kafka/subscriber.go index 0fe0791e7..45f9a3c0a 100644 --- a/message/infrastructure/kafka/subscriber.go +++ b/message/infrastructure/kafka/subscriber.go @@ -29,15 +29,25 @@ type confluentSubscriber struct { } type SubscriberConfig struct { + // Kafka brokers list. Brokers []string - ConsumerGroup string + // Kafka consumer group. + ConsumerGroup string + // When we want to consume without consumer group, you should set it to true. + // In practice you will receive all messages sent to the topic. NoConsumerGroup bool + // Action to take when there is no initial offset in offset store or the desired offset is out of range. + // Available options: smallest, earliest, beginning, largest, latest, end, error. AutoOffsetReset string + // How much consumers should be spawned. + // Every consumer will receive messages for their own partition so messages order will be preserved. ConsumersCount int + // Passing librdkafka options. + // Available options: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md KafkaConfigOverwrite kafka.ConfigMap } @@ -130,6 +140,9 @@ func DefaultConfluentConsumerConstructor(config SubscriberConfig) (*kafka.Consum return kafka.NewConsumer(kafkaConfig) } +// Subscribe subscribers for messages in Kafka. +// +// They are multiple subscribers spawned func (s *confluentSubscriber) Subscribe(topic string) (chan *message.Message, error) { if s.closed { return nil, errors.New("subscriber closed") diff --git a/message/message.go b/message/message.go index f924a5d23..8b087e6dd 100644 --- a/message/message.go +++ b/message/message.go @@ -20,16 +20,28 @@ var ( type Payload []byte type Message struct { - UUID string // todo - change to []byte?, change to type - + // UUID is an unique identifier of message. + // + // It is only used by Watermill for debugging. + // UUID can be empty. + UUID string + + // Metadata contains the message metadata. + // + // Can be used to store data which doesn't require unmarshaling entire payload. + // It is something similar to HTTP request's headers. Metadata Metadata + // Payload is message's payload. Payload Payload - ack chan struct{} - noAck chan struct{} - ackMutex sync.Mutex - ackSent ackType + // ack is closed, when acknowledge is received. + ack chan struct{} + // noACk is closed, when negative acknowledge is received. + noAck chan struct{} + + ackMutex sync.Mutex + ackSentType ackType } func NewMessage(uuid string, payload Payload) *Message { @@ -50,18 +62,23 @@ const ( nack ) +// Ack sends message's acknowledgement. +// +// Ack is not blocking. +// Ack is idempotent. +// Error is returned, if Nack is already sent. func (m *Message) Ack() error { m.ackMutex.Lock() defer m.ackMutex.Unlock() - if m.ackSent == nack { + if m.ackSentType == nack { return ErrAlreadyNacked } - if m.ackSent != noAckSent { + if m.ackSentType != noAckSent { return nil } - m.ackSent = ack + m.ackSentType = ack if m.noAck == nil { m.ack = closedchan } else { @@ -71,18 +88,23 @@ func (m *Message) Ack() error { return nil } +// Nack sends message's negative acknowledgement. +// +// Nack is not blocking. +// Nack is idempotent. +// Error is returned, if Ack is already sent. func (m *Message) Nack() error { m.ackMutex.Lock() defer m.ackMutex.Unlock() - if m.ackSent == ack { + if m.ackSentType == ack { return ErrAlreadyAcked } - if m.ackSent != noAckSent { + if m.ackSentType != noAckSent { return nil } - m.ackSent = nack + m.ackSentType = nack if m.noAck == nil { m.noAck = closedchan @@ -93,10 +115,28 @@ func (m *Message) Nack() error { return nil } +// Acked returns channel which is closed when acknowledgement is sent. +// +// Usage: +// select { +// case <-message.Acked(): +// // ack received +// case <-message.Nacked(): +// // nack received +// } func (m *Message) Acked() <-chan struct{} { return m.ack } +// Nacked returns channel which is closed when negative acknowledgement is sent. +// +// Usage: +// select { +// case <-message.Acked(): +// // ack received +// case <-message.Nacked(): +// // nack received +// } func (m *Message) Nacked() <-chan struct{} { return m.noAck } diff --git a/message/publisher.go b/message/publisher.go index bf3f40e25..c1f553628 100644 --- a/message/publisher.go +++ b/message/publisher.go @@ -1,10 +1,18 @@ package message type publisher interface { + // Publish publishes provided messages to given topic. + // + // Publish can be synchronous or asynchronous - it depends of implementation. + // + // Most publishers implementations doesn't support atomic publishing of messages. + // That means, that when publishing one of messages failed next messages will be not published. Publish(topic string, messages ...*Message) error } type Publisher interface { publisher + + // Close should flush unsent messages, if publisher is async. Close() error } diff --git a/message/pubsub.go b/message/pubsub.go index 17881a346..e9d87c6dc 100644 --- a/message/pubsub.go +++ b/message/pubsub.go @@ -9,6 +9,10 @@ type PubSub interface { Close() error } +func NewPubSub(publisher Publisher, subscriber Subscriber) PubSub { + return pubSub{publisher, subscriber} +} + type pubSub struct { Publisher Subscriber @@ -32,7 +36,3 @@ func (p pubSub) Close() error { return errors.New(errMsg) } - -func NewPubSub(publisher Publisher, subscriber Subscriber) PubSub { - return pubSub{publisher, subscriber} -} diff --git a/message/router.go b/message/router.go index 20fa0b4a1..7670ba0c6 100644 --- a/message/router.go +++ b/message/router.go @@ -10,13 +10,40 @@ import ( "github.com/pkg/errors" ) +// HandlerFunc is function called when message is received. +// +// msg.Ack() is called automatically when HandlerFunc doesn't return error. +// When HandlerFunc returns error, msg.Nack() is called. +// When msg.Ack() was called in handler and HandlerFunc returns error, +// msg.Nack() will be not sent because Ack was already sent. +// +// HandlerFunc's are executed parallel when multiple messages was received +// (because msg.Ack() was sent in HandlerFunc or Subscriber supports multiple consumers). type HandlerFunc func(msg *Message) ([]*Message, error) +// HandlerMiddleware allows us to write something like decorators to HandlerFunc. +// It can execute something before handler (for example: modify consumed message) +// or after (modify produced messages, ack/nack on consumed message, handle errors, logging, etc.). +// +// It can be attached to the router by using `AddMiddleware` method. +// +// Example: +// func ExampleMiddleware(h message.HandlerFunc) message.HandlerFunc { +// return func(message *message.Message) ([]*message.Message, error) { +// fmt.Println("executed before handler") +// producedMessages, err := h(message) +// fmt.Println("executed after handler") +// +// return producedMessages, err +// } +// } type HandlerMiddleware func(h HandlerFunc) HandlerFunc +// RouterPlugin is function which is executed on Router start. type RouterPlugin func(*Router) error type RouterConfig struct { + // CloseTimeout determines how long router should work for handlers when closing. CloseTimeout time.Duration } @@ -94,6 +121,20 @@ func (r *Router) AddPlugin(p ...RouterPlugin) { r.plugins = append(r.plugins, p...) } +// AddHandler adds a new handler. +// +// handlerName must be unique. For now, it is used only for debugging. +// +// subscribeTopic is a topic from which handler will receive messages. +// +// publishTopic is a topic to which router will produce messages retuened by handlerFunc. +// When handler needs to publish to multiple topics, +// it is recommended to just inject Publisher to Handler or implement middleware +// which will catch messages and publish to topic based on metadata for example. +// +// pubSub is PubSub from which messages will be consumed and to which created messages will be published. +// If you have separated Publisher and Subscriber object, +// you can create PubSub object by calling message.NewPubSub(publisher, subscriber). func (r *Router) AddHandler( handlerName string, subscribeTopic string, @@ -110,6 +151,15 @@ func (r *Router) AddHandler( return nil } +// AddNoPublisherHandler adds a new handler. +// This handler cannot return messages. +// When message is returned it will occur an error and Nack will be sent. +// +// handlerName must be unique. For now, it is used only for debugging. +// +// subscribeTopic is a topic from which handler will receive messages. +// +// subscriber is Subscriber from which messages will be consumed. func (r *Router) AddNoPublisherHandler( handlerName string, subscribeTopic string, @@ -138,6 +188,10 @@ func (r *Router) AddNoPublisherHandler( return nil } +// Run runs all plugins and handlers and starts subscribing to provided topics. +// This call is blocking until router is running. +// +// To stop Run() you should call Close() on the router. func (r *Router) Run() (err error) { if r.isRunning { return errors.New("router is already running") @@ -205,7 +259,10 @@ func (r *Router) Run() (err error) { // Running is closed when router is running. // In other words: you can wait till router is running using -// <- r.Running() +// fmt.Println("Starting router") +// go r.Run() +// <- r.Running() +// fmt.Println("Router is running") func (r *Router) Running() chan struct{} { return r.running } diff --git a/message/subscriber.go b/message/subscriber.go index b2324f5db..b3183f0b4 100644 --- a/message/subscriber.go +++ b/message/subscriber.go @@ -1,10 +1,17 @@ package message type subscriber interface { + // Subscribe returns output channel with messages from provided topic. + // Channel is closed, when Close() was called to the subscriber. + // + // To receive next message, `Ack()` must be called on the received message. + // If message processing was failed and message should be redelivered `Nack()` should be called. Subscribe(topic string) (chan *Message, error) } type Subscriber interface { subscriber + + // Close closes all subscriptions with their output channels and flush offsets etc. when needed. Close() error } diff --git a/netlify.toml b/netlify.toml new file mode 100644 index 000000000..61315ea79 --- /dev/null +++ b/netlify.toml @@ -0,0 +1,17 @@ +[build] + command = "hugo version && ./build.sh" + base = "docs/" + publish = "docs/public/" + +[context.production.environment] + HUGO_VERSION = "0.52" + HUGO_ENV = "production" + HUGO_ENABLEGITINFO = "true" + +[context.deploy-preview.environment] + HUGO_VERSION = "0.52" + HUGO_ENABLEGITINFO = "true" + +[context.branch-deploy.environment] + HUGO_VERSION = "0.52" + HUGO_ENABLEGITINFO = "true"