Watermill Concepts

The Watermill Model

Watermill is a Go library for building event-driven applications. Its design centers on a small set of composable abstractions:

• Publisher — the message.Publisher interface, with a single Publish(topic, ...*message.Message) method.
• Subscriber — the message.Subscriber interface, with a Subscribe(ctx, topic) method that returns a channel of *message.Message.
• Router — wires subscribers to handlers and handlers to publishers, managing the consume → process → publish lifecycle and acknowledgment.
• Middleware — cross-cutting wrappers around handlers (retry, recoverer, throttle, correlation ID, poison queue).
• CQRS — a higher-level component that builds command and event buses on top of any Publisher/Subscriber pair.

watermill-kubemq is a Pub/Sub plugin: it implements the message.Publisher and message.Subscriber interfaces backed by KubeMQ. Because Watermill's Router, middleware, and CQRS components depend only on those two interfaces, every Watermill component works unchanged — you swap in the KubeMQ Publisher and Subscriber and the rest of your Watermill code stays the same.

This page explains the plugin's model. For what each KubeMQ pattern guarantees at the broker level, link out to the core docs: Events, Events Store, Queues, and RPC.

The package import path is github.com/kubemq-io/watermill-kubemq/pkg/kubemq, imported throughout this section as kubemq.

One Instance, One Pattern

KubeMQ exposes several distinct messaging patterns. watermill-kubemq does not try to hide that distinction behind a single transport — instead, each Publisher and Subscriber is bound to exactly one pattern, selected through the Pattern field on its config. The PatternType enum has three values:

const (
    PatternEvents      PatternType = iota // KubeMQ fire-and-forget events
    PatternEventsStore                    // KubeMQ persistent events
    PatternQueues                         // KubeMQ reliable queues with ack/nack
)

You set the pattern when constructing the Publisher or Subscriber. The pattern is required and validated; an out-of-range value is rejected at construction time.

import kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"

pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
    Address: "localhost:50000",
    Pattern: kubemq.PatternEvents, // this Publisher serves Events only
    Logger:  logger,
})

sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
    Address: "localhost:50000",
    Pattern: kubemq.PatternEvents, // matching Subscriber
    Logger:  logger,
})

To use more than one pattern in the same application — for example Queues for commands and EventsStore for domain events — construct a separate Publisher/Subscriber per pattern, as the CQRS example below shows.

Pattern Semantics

Choosing a pattern is a delivery-guarantee decision. The three patterns differ in delivery semantics, acknowledgment, and persistence:

• Events require a subscriber to be active before the message is published; there is no acknowledgment and no persistence.
• EventsStore persists every event. The consumer offset auto-advances as messages are delivered, and a subscriber can replay history from a chosen starting point (see EventsStore start options).
• Queues hold each message until it is explicitly acknowledged. A negative acknowledgment returns the message to the queue for redelivery.

Marshaling

A Watermill *message.Message carries three things: a UUID, a Payload ([]byte), and Metadata (a map[string]string). A KubeMQ message carries a Body ([]byte) and Tags (a map[string]string). The marshaler maps between them.

The plugin defines three interfaces:

// Marshaler converts a Watermill message to a KubeMQ-compatible format.
type Marshaler interface {
    Marshal(topic string, msg *message.Message) (*MarshaledMessage, error)
}

// Unmarshaler converts a KubeMQ message to a Watermill message.
type Unmarshaler interface {
    Unmarshal(msg *ReceivedMessage) (*message.Message, error)
}

// MarshalerUnmarshaler combines both.
type MarshalerUnmarshaler interface {
    Marshaler
    Unmarshaler
}

DefaultMarshaler (used when you leave Marshaler/Unmarshaler unset) implements MarshalerUnmarshaler with a direct mapping:

• Watermill Payload → KubeMQ Body
• Watermill Metadata entries → KubeMQ Tags
• Watermill UUID → stored in the reserved tag _watermill_uuid

On the way back out, Unmarshal reconstructs the Watermill message: it reads the UUID from the _watermill_uuid tag (falling back to the KubeMQ message ID, then to a freshly generated UUID), copies all other tags into Metadata, and restores the body as the payload.

const WatermillUUIDTag = "_watermill_uuid"

func (m DefaultMarshaler) Marshal(topic string, msg *message.Message) (*MarshaledMessage, error) {
    tags := make(map[string]string, len(msg.Metadata)+1)
    tags[WatermillUUIDTag] = msg.UUID
    for k, v := range msg.Metadata {
        if k == WatermillUUIDTag {
            return nil, fmt.Errorf("watermill-kubemq: metadata key %q is reserved for Watermill UUID", WatermillUUIDTag)
        }
        tags[k] = v
    }
    return &MarshaledMessage{Body: msg.Payload, Tags: tags}, nil
}

To change the wire encoding — for example to compress the body or use a binary envelope — provide your own type implementing MarshalerUnmarshaler and set it on the config's Marshaler (Publisher) or Unmarshaler (Subscriber) field.

Ack/Nack Bridging

Watermill's processing model is acknowledgment-driven: after a handler runs, the Router calls msg.Ack() on success or msg.Nack() on failure. Bridging that onto KubeMQ depends on the pattern.

For Events and EventsStore, KubeMQ delivers without waiting for acknowledgment, so the plugin simply hands the message to the output channel — there is no broker round-trip on Ack()/Nack().

For Queues, each polled message must be settled explicitly. When the Subscriber delivers a queue message, it spawns a small bridge goroutine that waits on the Watermill message's settlement channels and translates them into KubeMQ calls:

go func(qm *kubemqSDK.QueueDownstreamMessage, wm *message.Message) {
    defer s.wg.Done()
    select {
    case <-wm.Acked():
        if err := qm.Ack(); err != nil {
            s.config.Logger.Error("Queue ack error", err, watermill.LogFields{ ... })
        }
    case <-wm.Nacked():
        if err := qm.Nack(); err != nil {
            s.config.Logger.Error("Queue nack error", err, watermill.LogFields{ ... })
        }
    case <-subCtx.Done():
        _ = qm.Nack()
    }
}(qMsg, wmMsg)

The mapping is:

• Watermill msg.Acked() → KubeMQ Ack() (removes the message from the queue)
• Watermill msg.Nacked() → KubeMQ Nack() (returns the message for redelivery)
• subCtx cancellation (subscriber closing or context cancelled) → KubeMQ Nack(), so an in-flight message is returned to the queue rather than lost

This is why retry-style middleware is only meaningful for Queues: a Nack triggers actual redelivery from the broker. For Events and EventsStore there is no redelivery channel.

Subscriber-Injected Metadata

When the Subscriber delivers a message, it adds KubeMQ-specific context to the Watermill Metadata so your handlers can inspect broker details. Events and EventsStore messages receive the following (Queues messages carry no injected metadata):

For EventsStore, the sequence and timestamp are taken from the received event and set alongside the channel:

wmMsg.Metadata.Set("_kubemq_sequence", fmt.Sprintf("%d", event.Sequence))
wmMsg.Metadata.Set("_kubemq_timestamp", event.Timestamp.Format(time.RFC3339Nano))
wmMsg.Metadata.Set("_kubemq_channel", event.Channel)

These keys are injected by the Subscriber, not produced by the publisher, so they appear only on received messages.

EventsStore Start Options

EventsStore is persistent, so a subscriber must declare where in the stored history it wants to begin. The plugin exposes this through EventsStoreStartOption and three companion fields, and maps each option to the corresponding kubemq-go/v2 subscription option:

The Subscriber resolves the option into a kubemq-go/v2 subscription option at subscribe time:

func (s *Subscriber) resolveEventsStoreStartOption() kubemqSDK.SubscriptionOption {
    switch s.config.EventsStoreStartOption {
    case StartFromFirst:
        return kubemqSDK.StartFromFirstEvent()
    case StartFromLast:
        return kubemqSDK.StartFromLastEvent()
    case StartFromSequence:
        return kubemqSDK.StartFromSequence(int(s.config.EventsStoreSequence))
    case StartFromTime:
        return kubemqSDK.StartFromTime(s.config.EventsStoreStartTime)
    case StartFromTimeDelta:
        return kubemqSDK.StartFromTimeDelta(s.config.EventsStoreTimeDelta)
    default:
        return kubemqSDK.StartFromNewEvents()
    }
}

Config validation enforces the field requirements: StartFromSequence needs EventsStoreSequence > 0, StartFromTime needs a non-zero EventsStoreStartTime, and StartFromTimeDelta needs EventsStoreTimeDelta > 0. A misconfigured start option fails at construction rather than silently falling back.

sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
    Address:                "localhost:50000",
    Pattern:                kubemq.PatternEventsStore,
    ConsumerGroup:          "my-group",
    EventsStoreStartOption: kubemq.StartFromFirst, // replay all stored events
    Logger:                 logger,
})

Streaming vs Non-Streaming

By default the Publisher opens a persistent streaming handle for its pattern at construction time and reuses it for every Publish call. This is the high-throughput path. Each pattern uses its own handle:

When DisableStreaming is set to true on PublisherConfig, the Publisher skips opening a stream and instead issues a synchronous send per Publish call. The streaming path keeps a long-lived connection open and is generally faster under sustained load; the synchronous path makes one request-scoped call and is simpler to reason about for low-volume or one-shot publishing.

// Streaming path (default): a stream handle is opened in NewPublisher
if !config.DisableStreaming {
    switch config.Pattern {
    case PatternEvents:
        handle, err := pub.client.SendEventStream(ctx)
        // ...
        pub.eventStream = handle
    case PatternEventsStore:
        handle, err := pub.client.SendEventStoreStream(ctx)
        // ...
        pub.eventStoreStream = handle
    case PatternQueues:
        handle, err := pub.client.QueueUpstream(ctx)
        // ...
        pub.queueUpstream = handle
    }
}

At publish time, the Publisher chooses the path based on whether a stream handle exists. For Events:

if p.eventStream != nil {
    // Streaming path (default)
    if err := p.eventStream.Send(event); err != nil {
        return fmt.Errorf("watermill-kubemq: send event error: %w", err)
    }
} else {
    // Non-streaming fallback (DisableStreaming: true)
    if err := p.client.SendEvent(msg.Context(), event); err != nil {
        return fmt.Errorf("watermill-kubemq: send event error: %w", err)
    }
}

The Subscriber side does not have a streaming toggle: Events and EventsStore use the SDK's push subscriptions, and Queues poll the broker with MaxItems and WaitTimeoutSeconds.

Connection Model

The plugin connects to KubeMQ over the kubemq-go/v2 gRPC client. There are two ways to obtain that client.

Owned client (from Address). When you supply an Address in host:port form, the Publisher/Subscriber creates its own client and owns its lifecycle — Close() on the Publisher or Subscriber also closes the client. The host and port are parsed from the address; if the port cannot be parsed, the plugin falls back to 50000, KubeMQ's default gRPC port.

func parsePort(address string) int {
    _, portStr, err := net.SplitHostPort(address)
    if err != nil {
        return 50000 // default gRPC port
    }
    port, err := strconv.Atoi(portStr)
    if err != nil {
        return 50000
    }
    return port
}

opts := []kubemqSDK.Option{
    kubemqSDK.WithAddress(parseHost(config.Address), parsePort(config.Address)),
}
// ClientID, AuthToken, TLS appended as options...
client, err := kubemqSDK.NewClient(context.Background(), opts...)
pub.client = client
pub.ownsClient = true

Reused client (ExistingClient). When you set ExistingClient, the Publisher/Subscriber uses that client directly and does not own it — Address, ClientID, AuthToken, and TLS are ignored, and the caller is responsible for closing the client. This lets a single gRPC connection back multiple Publishers and Subscribers (for example one per pattern).

OpenTelemetry Trace Propagation

The plugin propagates distributed-tracing context across the publish/subscribe boundary using the global OpenTelemetry TextMapPropagator and W3C Trace Context.

Before publishing, the Publisher injects the current trace context into the message metadata, so it travels through to KubeMQ Tags. After receiving, the Subscriber extracts that context back out and attaches it to the message:

func injectTraceContext(msg *message.Message) {
    otel.GetTextMapPropagator().Inject(
        msg.Context(),
        propagation.MapCarrier(msg.Metadata),
    )
}

func extractTraceContext(msg *message.Message) {
    ctx := otel.GetTextMapPropagator().Extract(
        msg.Context(),
        propagation.MapCarrier(msg.Metadata),
    )
    msg.SetContext(ctx)
}

Because the metadata becomes KubeMQ Tags (via the marshaler), the trace context survives the round trip. The kubemq-go/v2 SDK also has built-in OTel integration, and Watermill's community OpenTelemetry middleware can be layered on for handler-level spans.

Message Flow

The diagram below shows a Watermill Router wired to a KubeMQ Subscriber and Publisher, with the plugin translating between Watermill messages and KubeMQ messages on each side.

CQRS on KubeMQ

Because each Publisher/Subscriber is bound to one pattern, the natural CQRS layout uses Queues for commands (reliable, settled delivery) and EventsStore for domain events (persistent, replayable). You construct one pair per concern and hand them to Watermill's CQRS facade:

// Commands via Queues (reliable, exactly-once processing)
cmdPub, _ := kubemq.NewPublisher(kubemq.PublisherConfig{
    Address: "localhost:50000",
    Pattern: kubemq.PatternQueues,
})
cmdSub, _ := kubemq.NewSubscriber(kubemq.SubscriberConfig{
    Address: "localhost:50000",
    Pattern: kubemq.PatternQueues,
})

// Events via EventsStore (persistent, replayable)
evtPub, _ := kubemq.NewPublisher(kubemq.PublisherConfig{
    Address: "localhost:50000",
    Pattern: kubemq.PatternEventsStore,
})
evtSub, _ := kubemq.NewSubscriber(kubemq.SubscriberConfig{
    Address:                "localhost:50000",
    Pattern:                kubemq.PatternEventsStore,
    EventsStoreStartOption: kubemq.StartFromNew,
})

Running a broker

watermill-kubemq talks to KubeMQ over gRPC on port 50000, so all you need is a running broker — no HTTP connector flag is required for this plugin (the gRPC server is always on). For the Docker command and port breakdown, see Getting Started. Point your Publisher and Subscriber at localhost:50000 and you are ready to publish and subscribe.

Related topics