# Events (/integrations/watermill/how-to/events)



## Overview [#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](/learn/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).

<Mermaid
  chart="sequenceDiagram
    participant Pub as Publisher
    participant K as KubeMQ
    participant S1 as Subscriber 1
    participant S2 as Subscriber 2
    Pub->>K: Publish (fire-and-forget)
    K->>S1: deliver
    K->>S2: deliver
    Note over Pub,S2: No ack, no persistence.<br/>Subscribers must be connected first."
/>

<Callout type="info">
  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](/integrations/watermill/tutorials/getting-started).
</Callout>

## Publish and subscribe [#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.

```go title="events/basic-pubsub/main.go"
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.

<Callout type="warn">
  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.
</Callout>

## Fan-out vs consumer groups [#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.

```go title="events/consumer-group/main.go (excerpt)"
// 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.

<Callout type="info">
  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](/integrations/watermill/how-to/queues)
  pattern instead.
</Callout>

## Streaming publish [#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.

```go title="events/stream-publish/main.go (excerpt)"
// 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 [#run-the-examples]

With a broker running on `localhost:50000`, run the Events examples from the repository root:

```bash
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.go
```

## Next steps [#next-steps]

<Cards>
  <Card title="Events Store" href="/integrations/watermill/how-to/events-store" description="The persistent, replayable sibling of Events — durable storage, sequence numbers, and replay from any point." />

  <Card title="Queues" href="/integrations/watermill/how-to/queues" description="Reliable point-to-point delivery with explicit ack/nack, DLQ, and competing consumers." />

  <Card title="Reference" href="/integrations/watermill/reference/configuration" description="Full PublisherConfig and SubscriberConfig field reference." />
</Cards>