Events and event bus

eventbus.EventBus lets producers publish and walk away while zero or more live listeners react. Use it for notify-style fan-out where no reaction is load-bearing, like a booking.created signal. (The bookings example delivers booking.created through webhooks, not this bus.) No durability and no redelivery for offline handlers, so put must-run work on the queue via job.Dispatcher.

Quick start

EVENTBUS_ADAPTER=memory
EVENTBUS_BUFFER_SIZE=1024
EVENTBUS_MAX_HANDLERS=128
EVENTBUS_HANDLER_TIMEOUT=30s
EVENTBUS_CLOSE_TIMEOUT=5s

main.go

package main

import (
  "context"
  "fmt"

  eventbusmemory "github.com/zenta-dev/zever/adapters/eventbus/memory"
  "github.com/zenta-dev/zever/core/eventbus"
)

func main() {
  eventbusmemory.Register()
  bus, _ := eventbus.Open(eventbus.Memory, eventbus.Options{})
  defer bus.Close()
  ctx := context.Background()

  ch, _ := bus.SubscribeChan(ctx, "orders", 16)
  defer func() { _ = bus.Unsubscribe("orders", ch) }()

  unsub, _ := bus.Subscribe(ctx, "orders", func(_ context.Context, msg eventbus.Message) {
    fmt.Println("push saw", string(msg.Payload))
  })
  defer unsub()

  _ = bus.Publish(ctx, "orders", eventbus.NewPayload([]byte("hello")), nil)
  msg := <-ch
  fmt.Println("pull saw", string(msg.Payload))
}

Output: push saw hello and pull saw hello. Push (Subscribe) takes a handler and returns an unsubscribe closure; pull (SubscribeChan) returns a channel. Keep payloads small, versioned JSON.

Subscribe without leaking goroutines

Every subscription needs an explicit exit path, or leak tests flag the stray goroutine: defer the unsubscribe closure after Subscribe, and Unsubscribe after SubscribeChan. Closing the bus also ends subscriptions. Topics must be non-empty and at most 256 bytes (MaxTopicLen). Payloads are at most 4 MiB (MaxMessageSize); oversize publishes fail closed. Set OnPanic (never validated) where a bad subscriber must not take down the publisher. Cap fan-out with max_handlers: unbounded subscribes from request paths leak handlers.

Fan out across instances with Redis

Use redis when a publish on one process must reach a subscriber on another; call sites do not change.

YAML

EVENTBUS_ADAPTER=redis
EVENTBUS_BUFFER_SIZE=1024
EVENTBUS_REDIS_ADDR=localhost:6379
EVENTBUS_REDIS_DB=0
EVENTBUS_REDIS_PREFIX=bookings
EVENTBUS_REDIS_POOL_SIZE=20
EVENTBUS_REDIS_MIN_IDLE_CONNS=2
EVENTBUS_REDIS_POOL_TIMEOUT=5s

Env

EVENTBUS_ADAPTER=redis
EVENTBUS_BUFFER_SIZE=1024
EVENTBUS_HANDLER_TIMEOUT=30s

Keys are strict snake_case, with Redis pooling under a nested redis block. Zero means the adapter default (buffer_size 1024, max_handlers 128, handler_timeout 30s, close_timeout 5s); only negative values fail. Env is best-effort: unknown variables are ignored, unparsable resolved values error. All keys: eventbus config reference.

Adapter	Package	Notes
memory	adapters/eventbus/memory	Default, process-local, push plus pull, zero infra
redis	adapters/eventbus/redis	Shared fan-out, prefix-keyed, pool-backed

Resolve the bus from the container

main.go

eventbusmemory.Register()
bus, err := c.EventBus() // alias c.Eventbus()

EventBus() resolves lazily and caches; on error the cache clears. Open always returns a bus with pull support: adapters expose the Pusher push core and the package wraps it. Close() error takes no context. See the container.

Publishing to a topic with no subscriber is a quiet no-op.

Production checklist

• Not durable: a handler offline at publish time misses the message, with no replay.
• Backpressure drops newest: a full per-topic buffer (default 1024) drops the newest message for that subscriber only. Size buffer_size for burst plus handler latency.
• Bound handler time with handler_timeout (default 30s); a stuck handler holds its slot and its buffer fills, causing drops.
• Never put secrets in payloads. Redaction covers logged config, not message bodies.

Troubleshooting

A subscriber on host B never fires for publishes on host A

The memory adapter in a multi-instance deploy; delivery is in-process. Set eventbus.adapter to redis with a shared addr and prefix, then confirm every instance resolves the same endpoint through redacted output.

Messages vanish under load with no error

The per-topic buffer is full and drop-newest backpressure engaged. Raise buffer_size, shorten handler work (enqueue a job for slow I/O), and look for a slow subscriber starving its channel.

Strict load rejects buffersize or maxhandlers

Flat pre-rename spellings; keys are buffer_size, max_handlers, handler_timeout, close_timeout. Rename them and use integer nanoseconds. Env still accepts compact spellings because _ is ignored.

A leak test fails after adding a subscription

No exit path: the unsubscribe func was never called, or the SubscribeChan channel never reached Unsubscribe or bus Close. Defer both as above.

Related

QueueDurable delivery with ack and retry.

Scheduler and jobsDeferred execution instead of fan-out.

OutboxEvents that commit with the business row.