relay

package module
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 17, 2026 License: Apache-2.0 Imports: 7 Imported by: 0

README

Relay

Go Reference License Go Version

Zero-dependency, type-safe, context-aware event bus for Go.

Relay provides a simple Bus pattern for one-to-many communication using generics and Go channels. Built on the standard library with no external dependencies.

Features

  • Zero external dependencies. Only the Go standard library.
  • Fire-and-forget or wait. Broadcast() returns a BroadcastResult; ignore it for async delivery or call Wait() to block until all subscribers have been handled.
  • Configurable slow-subscriber policy. Block (default), drop, or apply a timeout per subscriber.
  • Per-subscription backpressure. Control the size of each subscriber's event buffer.
  • Context-aware. Subscriptions respect cancellation, and the bus can be shut down cleanly with Close().
  • Observability hooks. Attach a callback to monitor broadcast latency and drop rates.
  • Lock-free broadcasts. Subscriber snapshots use atomic.Pointer for zero contention on the hot path.
  • Sharded subscriber storage. Subscribe/unsubscribe scale with GOMAXPROCS shards.
  • Graceful shutdown. CloseGracefully waits for in-flight broadcasts.

Installation

go get -u github.com/binozo/go-relay

Requires Go 1.22 or later.

Quick Start

package main

import (
	"context"
	"fmt"

	"github.com/binozo/go-relay"
)

func main() {
	bus := relay.NewChanBus[string]()
	defer bus.Close()

	sub, err := bus.Subscribe()
	if err != nil {
		panic(err)
	}
	defer sub.Unsubscribe()

	go func() {
		for {
			msg, err := sub.Listen(context.TODO())
			if err != nil {
				return
			}
			fmt.Println("Received:", msg)
		}
	}()

	result := bus.Broadcast("Hello, Relay!")
	_ = result.Wait() // optional: wait until the event has been delivered
}

Configuration

bus := relay.NewChanBus[string](
    relay.WithPoolSize(4),               // limit concurrent broadcast goroutines; defaults to GOMAXPROCS
    relay.WithShardCount(8),              // subscriber list shards; defaults to GOMAXPROCS
    relay.WithSubscriptionBackpressure(10), // each subscriber gets a buffered channel of size 10
    relay.WithSlowSubscriberPolicy(relay.SlowSubscriberPolicyDrop), // drop events for slow readers
    relay.WithSlowSubscriberTimeout(100*time.Millisecond),           // only used with SlowSubscriberPolicyTimeout
    relay.WithBroadcastCallback(func(d time.Duration, alive, dropped int) {
        fmt.Printf("broadcast: %d alive, %d dropped, took %v\n", alive, dropped, d)
    }),
)

Observability & Metrics

// Query live subscriber count
fmt.Println("subscribers:", bus.Len())

// Query aggregate stats across all subscribers
agg := bus.Stats()
fmt.Printf("received=%d dropped=%d subscribers=%d\n", agg.EventsReceived, agg.EventsDropped, agg.Subscribers)

// Per-subscriber stats
stats := sub.Stats()
fmt.Printf("received=%d dropped=%d\n", stats.EventsReceived, stats.EventsDropped)

Backpressure

If WithPoolSize is configured, the semaphore caps concurrent broadcasts. Use TryBroadcast for non-blocking attempts:

result, ok := bus.TryBroadcast("urgent")
if !ok {
    // bus is saturated -- drop or retry later
}

For blocking semantics (wait until the semaphore has room), use Broadcast:

result := bus.Broadcast("normal")
if err := result.Wait(); errors.Is(err, relay.ErrBroadcastFull) {
    // only possible if poolSize > 0 and the bus is saturated
}

Error Handling

Method Error Meaning
Bus.Subscribe() relay.ErrBusClosed Bus has been shut down.
Bus.SubscribeContext(ctx) context.Canceled / context.DeadlineExceeded Context was cancelled before subscription completed.
Bus.Broadcast() - Returns a BroadcastResult; actual error comes from Wait().
BroadcastResult.Wait() relay.ErrBusClosed Bus closed during broadcast.
BroadcastResult.Wait() relay.ErrBroadcastFull Semaphore saturated (WithPoolSize only).
Subscription.Listen(ctx) relay.ErrSubscriptionClosed Subscription unsubscribed.
Subscription.Listen(ctx) ctx error Listener context was cancelled.

Context-Aware Subscription

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

sub, err := bus.SubscribeContext(ctx)
if err != nil {
    // handle error
}
defer sub.Unsubscribe()

// If ctx expires, Listen will return the context error
msg, err := sub.Listen(context.Background())

Graceful Shutdown

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

if err := bus.CloseGracefully(ctx); err != nil {
    // timeout -- some broadcasts were still in flight
}

API Reference

Type Description
ChanBus[T] Main event bus for type T.
BroadcastResult Handle for tracking a broadcast; call Wait() to block.
Subscription[T] Receives events of type T via Listen(ctx).
Stats Aggregated subscriber metrics.
SlowSubscriberPolicy Block / Drop / Timeout strategy.

Full API docs: pkg.go.dev

Benchmarks

Run go test -bench=. -benchmem . in the project root.

Environment: Go 1.22, AMD Ryzen 5 5600X, Linux

Broadcast Throughput
Subscribers ops/sec ns/op B/op allocs/op
1 ~270k 3,709 672 9
10 ~82k 14,278 4,128 54
100 ~18k 64,305 38,708 504
1000 ~2.4k 497,939 384,609 5,006
Slow-Subscriber Policy (100 subscribers, fast readers)
Policy ops/sec ns/op B/op allocs/op
Block ~39k 31,228 288 4
Drop ~40k 29,698 288 4
Timeout (1ms) ~25k 48,065 536 7
Churn (subscribe -> broadcast -> unsubscribe)
Shards ops/sec ns/op B/op allocs/op
1 ~521k 2,031 707 16
4 ~567k 2,039 732 16
12 ~465k 2,150 795 16
Utilities
Operation ops/sec ns/op B/op allocs/op
TryBroadcast ~5.8M 194 97 1
Stats (10 subs) ~6.1M 193 80 1
Stats (100 subs) ~1.2M 1,015 896 1
Stats (1000 subs) ~125k 9,828 8,194 1
Notes on Benchmarks
  • Broadcast cost scales linearly with subscriber count by design: one sequential goroutine iterates the snapshot. The broadcast hot path itself allocates 3-5 times regardless of subscriber count. Higher numbers in BenchmarkBroadcast come from the benchmark's background listener goroutines, which create a context.WithTimeout on every Listen() call. This is not overhead from Relay.
  • TryBroadcast is sub-microsecond.
  • Churn throughput is ~500k/sec regardless of shard count because the dominant cost is goroutine creation, not shard lookup.

Design

Relay is intentionally simple:

  1. Snapshots, not locks. On every broadcast, the bus takes an atomic snapshot of the current subscriber list. Readers never block writers and vice versa.
  2. Sharded maps. Subscribers are distributed across GOMAXPROCS shards to reduce contention during subscribe/unsubscribe.
  3. Channels everywhere. Events flow through Go channels for familiarity and composability with standard patterns.

License

Apache 2.0

Documentation

Overview

Package relay provides a high-throughput, type-safe pub/sub event bus with sharded subscriber storage, configurable backpressure, and graceful shutdown.

Create a bus with NewChanBus, subscribe with Subscribe or SubscribeContext, and broadcast events with Broadcast or TryBroadcast.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrUnsubscribed  = errors.New("subscription closed")
	ErrBusClosed     = errors.New("bus closed")
	ErrBroadcastFull = errors.New("broadcast semaphore full")
)

Functions

This section is empty.

Types

type AggregateStats 

type AggregateStats struct {
	Subscribers    int
	EventsReceived uint64
	EventsDropped  uint64
}

AggregateStats holds bus-wide delivery metrics across all active subscriptions.

type BroadcastResult 

type BroadcastResult struct {
	// contains filtered or unexported fields
}

BroadcastResult allows waiting for a broadcast to finish.

func (BroadcastResult) Wait 

func (b BroadcastResult) Wait() error

Wait blocks until the broadcast completes and returns an error if the bus was closed before or during the broadcast (ErrBusClosed) or if the broadcast was rejected due to a full semaphore (ErrBroadcastFull).

func (BroadcastResult) WaitContext 

func (b BroadcastResult) WaitContext(ctx context.Context) error

WaitContext blocks until the broadcast completes or ctx is cancelled. It may return ErrBusClosed or ErrBroadcastFull.

type Bus 

type Bus[T any] interface {
	// Broadcast sends an event to all active subscribers. It returns
	// immediately; use BroadcastResult.Wait() to block until delivery is
	// complete. If poolSize > 0, Broadcast blocks until the semaphore has room.
	Broadcast(event T) BroadcastResult

	// TryBroadcast attempts to broadcast without blocking. It returns (result,
	// true) on success, or (result, false) if the semaphore is saturated.
	// The result's Wait() will return ErrBroadcastFull.
	TryBroadcast(event T) (BroadcastResult, bool)

	// Subscribe creates a new subscription with the bus defaults. Additional
	// ChanSubscriptionOption values may override those defaults.
	Subscribe(options ...ChanSubscriptionOption) (Subscription[T], error)

	// SubscribeContext creates a subscription tied to the given context. When
	// ctx is cancelled, the subscription is automatically unsubscribed.
	SubscribeContext(ctx context.Context, options ...ChanSubscriptionOption) (Subscription[T], error)

	// Len returns the total number of subscribers currently connected to the bus.
	Len() int

	// Stats returns aggregate delivery statistics across all active subscriptions.
	Stats() AggregateStats

	// Close force-closes the bus immediately. In-flight broadcasts are not
	// waited for.
	Close() error

	// CloseGracefully rejects new operations, waits for in-flight broadcasts
	// to finish (or until ctx is cancelled), and then closes the bus.
	CloseGracefully(ctx context.Context) error
}

Bus[T] is a type-safe event broadcaster. Implementations must be safe for concurrent use by multiple goroutines.

type ChanBus 

type ChanBus[T any] struct {
	// contains filtered or unexported fields
}

func NewChanBus 

func NewChanBus[T any](options ...ChanBusOption) *ChanBus[T]

NewChanBus creates a new event bus with the provided options.

func (*ChanBus[T]) Broadcast 

func (c *ChanBus[T]) Broadcast(event T) BroadcastResult

Broadcast sends the event to all active subscribers. It blocks until the semaphore has room (if poolSize is configured), then returns immediately; use BroadcastResult.Wait() to block until the fan-out is complete.

func (*ChanBus[T]) Close 

func (c *ChanBus[T]) Close() error

func (*ChanBus[T]) CloseGracefully 

func (c *ChanBus[T]) CloseGracefully(ctx context.Context) error

CloseGracefully initiates a graceful shutdown. It rejects new subscriptions and broadcasts, waits for all in-flight broadcasts to complete (or until ctx is cancelled), and then closes the bus. If ctx expires before all broadcasts finish, the bus is force-closed and ctx.Err() is returned.

func (*ChanBus[T]) Len 

func (c *ChanBus[T]) Len() int

Len returns the total number of subscribers currently connected to the bus.

func (*ChanBus[T]) Stats 

func (c *ChanBus[T]) Stats() AggregateStats

Stats returns aggregate statistics across all active subscriptions.

func (*ChanBus[T]) Subscribe 

func (c *ChanBus[T]) Subscribe(options ...ChanSubscriptionOption) (Subscription[T], error)

func (*ChanBus[T]) SubscribeContext 

func (c *ChanBus[T]) SubscribeContext(ctx context.Context, options ...ChanSubscriptionOption) (Subscription[T], error)

SubscribeContext creates a new subscription attached to ctx.

func (*ChanBus[T]) TryBroadcast 

func (c *ChanBus[T]) TryBroadcast(event T) (BroadcastResult, bool)

TryBroadcast attempts to broadcast immediately. It returns the result and true on success, or false if the broadcast semaphore is saturated.

type ChanBusOption 

type ChanBusOption func(cfg *busConfig)

ChanBusOption configures a ChanBus during construction.

func WithBroadcastCallback 

func WithBroadcastCallback(cb func(duration time.Duration, alive, dropped int)) ChanBusOption

WithBroadcastCallback attaches a callback that is invoked after every broadcast completes. It receives the broadcast duration, the number of active (alive) subscribers, and how many events were dropped.

func WithContext 

func WithContext(ctx context.Context) ChanBusOption

WithContext sets the parent context for the bus. When this context is cancelled, all operations (Subscribe, Broadcast) will fail with ErrBusClosed.

func WithPoolSize 

func WithPoolSize(poolSize int) ChanBusOption

WithPoolSize caps the number of concurrent broadcast goroutines. A value of 0 means unlimited; values > 0 create a semaphore of that size. If the semaphore is full, Broadcast blocks until room is available, while TryBroadcast returns ErrBroadcastFull. Defaults to runtime.GOMAXPROCS(0).

func WithShardCount 

func WithShardCount(shardCount int) ChanBusOption

WithShardCount sets the number of subscriber shards. Higher values reduce subscribe/unsubscribe contention but increase the cost of snapshots. Defaults to runtime.GOMAXPROCS(0).

func WithSlowSubscriberPolicy 

func WithSlowSubscriberPolicy(policy SlowSubscriberPolicy) ChanBusOption

WithSlowSubscriberPolicy selects how the bus treats a subscriber whose buffer is full. Block waits, Drop skips immediately, Timeout gives up after WithSlowSubscriberTimeout. Defaults to Block.

func WithSlowSubscriberTimeout 

func WithSlowSubscriberTimeout(timeout time.Duration) ChanBusOption

WithSlowSubscriberTimeout sets the per-subscriber timeout used by the Timeout policy. Has no effect on Block or Drop policies.

func WithSubscriptionBackpressure 

func WithSubscriptionBackpressure(backpressure int) ChanBusOption

WithSubscriptionBackpressure sets the default buffer size for every subscriber created by this bus. Individual subscriptions may override this with SubWithBackpressure. A value of 0 creates an unbuffered channel.

type ChanSubscription 

type ChanSubscription[T any] struct {
	// contains filtered or unexported fields
}

func NewChanSubscription 

func NewChanSubscription[T any](options ...ChanSubscriptionOption) *ChanSubscription[T]

func (*ChanSubscription[T]) Close 

func (c *ChanSubscription[T]) Close() error

func (*ChanSubscription[T]) Listen 

func (c *ChanSubscription[T]) Listen(ctx context.Context) (T, error)

func (*ChanSubscription[T]) Stats 

func (c *ChanSubscription[T]) Stats() SubscriptionStats

func (*ChanSubscription[T]) String 

func (c *ChanSubscription[T]) String() string

func (*ChanSubscription[T]) Unsubscribe 

func (c *ChanSubscription[T]) Unsubscribe()

type ChanSubscriptionOption 

type ChanSubscriptionOption func(cfg *chanSubscriptionConfig)

ChanSubscriptionOption configures an individual subscription during creation.

func SubWithBackpressure 

func SubWithBackpressure(backpressure int) ChanSubscriptionOption

SubWithBackpressure sets the size of the subscription's internal buffer. A value of 0 creates an unbuffered channel.

func SubWithContext 

func SubWithContext(ctx context.Context) ChanSubscriptionOption

SubWithContext sets the parent context for the subscription. When this context is cancelled, Listen returns the context error.

func SubWithUnsubscribeCallback 

func SubWithUnsubscribeCallback(unsubscribeCallback func()) ChanSubscriptionOption

SubWithUnsubscribeCallback registers a function that is called when the subscription is unsubscribed. The bus also installs its own internal callback to remove the subscription from the shard; both are executed.

type Shards 

type Shards[T any] struct {
	// contains filtered or unexported fields
}

Shards holds subscriber lists partitioned across K shards.

func NewShards 

func NewShards[T any](count int) *Shards[T]

func (*Shards[T]) All 

func (s *Shards[T]) All() []*ChanSubscription[T]

All returns every subscription in every shard.

func (*Shards[T]) Cleanup 

func (s *Shards[T]) Cleanup()

Cleanup removes dead subscriptions from every shard.

func (*Shards[T]) Len 

func (s *Shards[T]) Len() int

Len returns the total number of subscriptions across all shards.

func (*Shards[T]) Snapshot 

func (s *Shards[T]) Snapshot() []*subList[T]

Snapshot returns a copy of all current shard lists.

func (*Shards[T]) Subscribe 

func (s *Shards[T]) Subscribe(sub *ChanSubscription[T]) int

Subscribe places sub into the next shard by round-robin. On success it returns the shard index that was chosen.

func (*Shards[T]) Unsubscribe 

func (s *Shards[T]) Unsubscribe(sub *ChanSubscription[T])

Unsubscribe scans all shards when the shard index is unknown. Prefer UnsubscribeAt.

func (*Shards[T]) UnsubscribeAt 

func (s *Shards[T]) UnsubscribeAt(idx int, sub *ChanSubscription[T])

UnsubscribeAt removes sub from shard idx only.

type SlowSubscriberPolicy 

type SlowSubscriberPolicy int

SlowSubscriberPolicy controls how the bus behaves when a subscriber's channel is full or the subscriber is otherwise slow to consume events. 

const (
	// SlowSubscriberPolicyBlock waits until the subscriber has room in its
	// buffer or until the subscriber/bus context is cancelled.
	SlowSubscriberPolicyBlock SlowSubscriberPolicy = iota
	// SlowSubscriberPolicyDrop skips the subscriber immediately if its buffer
	// is full. No event is lost for other subscribers.
	SlowSubscriberPolicyDrop
	// SlowSubscriberPolicyTimeout attempts delivery for up to the configured
	// duration (see WithSlowSubscriberTimeout) before skipping the subscriber.
	SlowSubscriberPolicyTimeout
)

type Subscription 

type Subscription[T any] interface {
	Listen(ctx context.Context) (T, error)
	Unsubscribe()
	Stats() SubscriptionStats
	String() string
	Close() error
}

Subscription[T] represents a single consumer attached to a Bus.

type SubscriptionStats 

type SubscriptionStats struct {
	EventsReceived uint64
	EventsDropped  uint64
}

SubscriptionStats holds per-subscription delivery metrics.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.