aio-watermill-kit

Watermill — Event-Driven Go Applications

Watermill is a Go library for building event-driven applications. It provides a clean abstraction over message brokers with a three-tier API: low-level Publisher/Subscriber interfaces, a mid-level Router with middleware support, and a high-level CQRS component.

The library's core philosophy: at-least-once delivery with idempotent handlers. Every handler you write should safely handle receiving the same message twice.

Architecture Overview

┌─────────────────────────────────────────────────┐
│  CQRS (High-level)                              │
│  CommandBus, EventBus, Processors, Marshalers   │
├─────────────────────────────────────────────────┤
│  Router (Mid-level)                             │
│  Handlers, Middleware chains, Lifecycle mgmt    │
├─────────────────────────────────────────────────┤
│  Publisher / Subscriber (Low-level)             │
│  Kafka, AMQP, Redis, NATS, SQL, GoChannel...   │
└─────────────────────────────────────────────────┘

Choose the right tier:

• Publisher/Subscriber — when you need raw control or simple publish/consume
• Router — when you need middleware, multiple handlers, or message transformation (most common)
• CQRS — when you want typed commands/events with automatic marshaling

Core Types

Message

The fundamental data unit. Immutable after publishing.

msg := message.NewMessage(watermill.NewUUID(), payload)
msg.Metadata.Set("correlation_id", corrID)

// Lifecycle: created → published → acked/nacked
// Router handles Ack/Nack automatically — don't call them manually in handlers

Fields:

• UUID (string) — unique ID for tracing/dedup, use watermill.NewUUID() or watermill.NewULID()
• Payload ([]byte) — message body, any format (JSON, Protobuf, etc.)
• Metadata (message.Metadata) — key-value headers, correlation IDs, partition keys
• Context — standard context.Context for cancellation and deadlines

Publisher & Subscriber