Getting Started with Watermill

Prerequisites

You need the following installed locally:

The Go version requirement comes from the plugin's module (go 1.25.0); the runnable examples in the repository assume the same toolchain.

Start a KubeMQ Broker

The plugin connects to KubeMQ over gRPC, so you need a broker running on your machine. The fastest way is the docker-compose.yml shipped with the repository:

docker-compose up -d

If you prefer to run the container directly without Compose:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 8080:8080 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

The broker exposes three ports:

Install the Plugin

Add the module to your project with go get:

go get github.com/kubemq-io/watermill-kubemq

This pulls in the plugin along with its dependencies, including Watermill (github.com/ThreeDotsLabs/watermill) and the KubeMQ Go SDK (github.com/kubemq-io/kubemq-go/v2).

Run Your First End-to-End Example

The example below wires a Watermill Router that reads from one Events topic, logs each message, and forwards it to an output topic — then publishes hello events into the input topic. Save it as main.go and run it with a broker listening on localhost:50000.

package main

import (
    "context"
    "log"

    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill/message"
    "github.com/ThreeDotsLabs/watermill/message/router/middleware"
    kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)

func main() {
    logger := watermill.NewStdLogger(false, false)

    pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
        Address: "localhost:50000",
        Pattern: kubemq.PatternEvents,
        Logger:  logger,
    })
    if err != nil {
        log.Fatal(err)
    }
    defer pub.Close()

    sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
        Address: "localhost:50000",
        Pattern: kubemq.PatternEvents,
        Logger:  logger,
    })
    if err != nil {
        log.Fatal(err)
    }
    defer sub.Close()

    router, err := message.NewRouter(message.RouterConfig{}, logger)
    if err != nil {
        log.Fatal(err)
    }

    router.AddMiddleware(middleware.Recoverer)
    router.AddHandler("events-handler", "my-topic", sub, "output-topic", pub,
        func(msg *message.Message) ([]*message.Message, error) {
            log.Printf("Received: %s", string(msg.Payload))
            return []*message.Message{msg}, nil
        },
    )

    // Publish a message
    go func() {
        msg := message.NewMessage(watermill.NewUUID(), []byte("hello events"))
        if err := pub.Publish("my-topic", msg); err != nil {
            log.Printf("Publish error: %v", err)
        }
    }()

    _ = router.Run(context.Background())
}

Run it from your module:

go run main.go

You should see the router log Received: hello events as the message flows through the handler.

Understand the Pieces

A few things are worth calling out in the example above.

The imports. The plugin builds on Watermill's core packages plus the KubeMQ adapter:

The pattern. Both the publisher and subscriber are created with Pattern: kubemq.PatternEvents. KubeMQ supports three messaging patterns, and each Publisher/Subscriber instance serves exactly one. PatternEvents is the fire-and-forget pattern.

The router. message.NewRouter builds a router; AddMiddleware(middleware.Recoverer) wraps handlers so a panic is recovered instead of crashing the process; and AddHandler wires an input topic + subscriber to an output topic + publisher with a handler function. Returning the message from the handler forwards it to output-topic.

Try the Runnable Examples

The repository ships complete, runnable programs for the Events pattern. With a broker running on localhost:50000, run them from the project root:

go run ./examples/events/basic-pubsub/main.go

The basic-pubsub example uses the direct Publisher/Subscriber API to send and receive three messages on watermill-events.basic-pubsub. For the Router-based flow shown above, run:

go run ./examples/events/router-handler/main.go

The router-handler example reads from watermill-events.router-input, uppercases each payload in the handler, and forwards it to watermill-events.router-output.

Next Steps