Events
Publish and subscribe with the fire-and-forget Events pattern — fan-out, consumer groups, and streaming publish over the Watermill interfaces.
Overview
The Events pattern (PatternEvents) maps Watermill's message.Publisher and
message.Subscriber onto KubeMQ's fire-and-forget events. Delivery is at-most-once:
there is no acknowledgment and nothing is persisted. A published message is fanned out to
whichever subscribers are connected at that instant, then it is gone — ideal for live
metrics, telemetry, log streams, and UI notifications where a missed message is acceptable.
For what the Events pattern guarantees at the broker level, see Events. This page documents how the plugin exposes it through the Watermill interfaces.
A publisher and subscriber are both created with Pattern: kubemq.PatternEvents. Because
delivery is fire-and-forget, the subscriber must be running before the publish happens
— if no subscriber is attached when you publish, the message is silently dropped (no
buffer, no replay).
The Watermill plugin is a native gRPC client — it dials the broker's gRPC port (50000)
directly, with no HTTP connector or enable flag to configure. If you have not set up a
broker yet, see Getting Started.
Publish and subscribe
A Publisher and Subscriber are created with PatternEvents and a Watermill Logger.
You Subscribe first, give the subscription a moment to establish, then Publish to the
same topic. Watermill's Subscribe returns a <-chan *message.Message that you range over.
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/ThreeDotsLabs/watermill"
"github.com/ThreeDotsLabs/watermill/message"
kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)
func main() {
logger := watermill.NewStdLogger(false, false)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// Create publisher
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternEvents,
Logger: logger,
})
if err != nil {
log.Fatal(err)
}
defer pub.Close()
// Create subscriber
sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternEvents,
Logger: logger,
})
if err != nil {
log.Fatal(err)
}
defer sub.Close()
// Subscribe to topic -- must be active before publish for events (fire-and-forget)
msgs, err := sub.Subscribe(ctx, "watermill-events.basic-pubsub")
if err != nil {
log.Fatal(err)
}
// Allow subscription to establish before publishing
time.Sleep(time.Second)
// Publish 3 messages
for i := 1; i <= 3; i++ {
msg := message.NewMessage(watermill.NewUUID(), []byte(fmt.Sprintf("Event message %d", i)))
if err := pub.Publish("watermill-events.basic-pubsub", msg); err != nil {
log.Fatal(err)
}
fmt.Printf("Published: %s\n", string(msg.Payload))
}
// Receive messages from subscription channel
received := 0
for received < 3 {
select {
case msg := <-msgs:
fmt.Printf("Received: UUID=%s, Payload=%s\n", msg.UUID, string(msg.Payload))
// Events don't require ack, but we demonstrate the Watermill pattern
msg.Ack()
received++
case <-ctx.Done():
log.Fatal("Timeout waiting for messages")
}
}
fmt.Println("Done! All 3 messages received.")
}Calling msg.Ack() on an Events message is harmless and keeps your handler code uniform
across patterns, but for Events it is a no-op at the broker — there is nothing to settle.
Because Events is fire-and-forget, the Subscribe call must complete (and the subscription
must be registered on the broker) before you publish. The time.Sleep above is a simple way
to demonstrate this in a single-process example; in a real deployment your subscribers are
long-lived and already connected.
Fan-out vs consumer groups
How an Events message is distributed depends entirely on whether subscribers set a
ConsumerGroup:
ConsumerGroup | Behavior | Example |
|---|---|---|
empty ("") | Fan-out — every subscriber gets every message | events/multiple-subscribers |
| shared name | Load-balanced — each message goes to one group member | events/consumer-group |
Internally the subscriber passes ConsumerGroup straight to KubeMQ's SubscribeToEvents,
so the grouping is enforced by the broker, not the plugin.
// Create 3 subscribers in the SAME consumer group.
// The 9 published messages are distributed across them (load-balanced),
// instead of every subscriber receiving all 9.
for i := 0; i < 3; i++ {
sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternEvents,
ConsumerGroup: "event-workers", // <-- shared group => competing consumers
Logger: logger,
})
if err != nil {
log.Fatal(err)
}
defer sub.Close()
msgs, err := sub.Subscribe(ctx, "watermill-events.consumer-group")
if err != nil {
log.Fatal(err)
}
channels[i] = msgs
}Omit ConsumerGroup (leave it "") and the same three subscribers each receive all nine
messages instead — that is the fan-out behavior shown in the events/multiple-subscribers
example.
Load-balancing of Events across a consumer group depends on broker support for event groups. If your KubeMQ server does not support it, grouped subscribers fall back to receiving all messages. For guaranteed work distribution with acknowledgment, use the Queues pattern instead.
Streaming publish
By default the Events publisher opens a persistent gRPC streaming handle
(SendEventStream) at construction time and sends every Publish call over that single
long-lived stream — the high-throughput path. Set DisableStreaming: true on the
PublisherConfig to fall back to a one-call-per-message synchronous send (SendEvent),
trading throughput for simpler, request-scoped error handling.
// Streaming publisher (default) -- sends over a persistent gRPC stream.
streamPub, _ := kubemq.NewPublisher(kubemq.PublisherConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternEvents,
Logger: logger,
})
// Non-streaming publisher -- one gRPC call per message.
nonStreamPub, _ := kubemq.NewPublisher(kubemq.PublisherConfig{
Address: "localhost:50000",
Pattern: kubemq.PatternEvents,
DisableStreaming: true,
Logger: logger,
})For most workloads, leave streaming enabled. Reach for DisableStreaming only when you have
a specific reason to send synchronously.
Run the examples
With a broker running on localhost:50000, run the Events examples from the repository root:
go run ./examples/events/basic-pubsub/main.go
go run ./examples/events/consumer-group/main.go
go run ./examples/events/multiple-subscribers/main.go
go run ./examples/events/router-handler/main.go
go run ./examples/events/stream-publish/main.goNext steps
Was this page helpful?