bus

package module
v0.11.1 Latest Latest
Warning

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

 Go to latest Published: Jul 31, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

README

Townbell Bus

Townbell Bus

A dependency-free, type-safe event bus for Go applications.

CI Go Reference MIT License

bus is for in-process event dispatch: no broker, no runtime dependency, and no serialization layer. It gives a Go application type-safe fan-out, bounded asynchronous work, priority, filters, middleware, and observable failures.

中文文档 · API reference · Examples · Migration from v0.5.x

API preview. Since v0.6.0, subscriptions use one Subscribe(topic, handler, options...) method, handlers return error, and Publish returns synchronous delivery failures. This API is intended to freeze at v1.0.0; feedback is welcome before then.

Why Townbell?

Need What you get
Keep one process decoupled Typed topics, fan-out, exact and hierarchical topic patterns
Move slow work off the request path Async handlers with serial or bounded-concurrency execution
Make failures visible Joined publish errors, per-error collection, panic recovery, error hooks, metrics
Stop safely Context-aware handlers, WaitAsync, and graceful Close

If an event must survive a process restart or cross machine boundaries, use a durable broker. Townbell deliberately stays on the lightweight side of that boundary.

Dispatch model

flowchart LR
    P["Publisher"] --> D["Publish(topic, event)"]
    D --> M["Middleware chain"]
    M --> X["Match exact + pattern handlers\n* / orders.*"]
    X --> H["Priority, filter, and once rules"]
    H --> S["Synchronous handler\nreturns to publisher"]
    H --> A["Async handler\nserial or max N"]
    S --> R["Publish error / PublishCollect"]
    A --> E["ErrorHandler + metrics"]

Install

go get github.com/townbell/bus

The core module only imports the Go standard library. Prometheus support is optional and lives in its own module:

go get github.com/townbell/bus/prometheus

Quick start

package main

import (
    "context"
    "fmt"

    "github.com/townbell/bus"
)

type UserEvent struct {
    ID     string
    Action string
}

func main() {
    b := bus.NewTyped[UserEvent]()
    defer b.Close()

    handle, err := b.Subscribe("user.login", func(ctx context.Context, event UserEvent) error {
        fmt.Printf("%s: %s\n", event.ID, event.Action)
        return nil
    })
    if err != nil {
        panic(err)
    }
    defer handle.Unsubscribe()

    if err := b.Publish("user.login", UserEvent{ID: "u-1", Action: "login"}); err != nil {
        fmt.Println("delivery failed:", err)
    }
}

Pick a path

I want to… Start here It demonstrates
Learn the API basic_example.go Typed subscriptions, options, async delivery
Configure dispatch advanced_usage.go Priority, filters, context, timeout, metrics
Instrument every event middleware_example.go Middleware ordering and interception
Publish from HTTP http_example.go Request contexts, patterns, dead events, shutdown
Run local background jobs worker_example.go Async work, HandlerMaxConcurrency, error hooks, draining
Follow a larger flow e_commerce_example.go Domain events, priorities, compensation

Run any example from its directory:

cd example && go run worker_example.go

Delivery semantics

Concern Contract
Publish Runs synchronous handlers in priority order and returns their joined errors. Later handlers still run after an ordinary error.
PublishCollect Returns every synchronous failure in dispatch order when callers need to retry, classify, or log separately.
Async handlers Return before the handler finishes; failures are reported only through ErrorHandler. Use async to free the caller, not to make CPU work faster.
Patterns * matches every topic. orders.* matches orders.created and deeper descendants, but not orders.
Context and timeout Cancellation stops later synchronous dispatch and is passed to the current handler. Handlers must honor their context to stop promptly.
Shutdown Close rejects new publish/subscribe calls and waits for already-started async work. Call WaitAsync when a process must drain earlier.

Useful options compose in one subscription:

b.Subscribe("payment.validate", validate,
    bus.HandlerPriority(bus.PriorityHigh),
    bus.HandlerTimeout(2*time.Second),
    bus.HandlerRecoverPolicy(bus.RecoverAndStop),
    bus.HandlerMaxConcurrency(4),
)

Available options: HandlerPriority, HandlerFilter, HandlerContext, HandlerAsync, HandlerOnce, HandlerTimeout, HandlerRecoverPolicy, HandlerMaxConcurrency, and HandlerSerial.

Features at a glance

Area Included
Routing Exact topics, *, hierarchical prefix.*, dead-event hook
Execution Sync/async handlers, priority, once, filter, context, timeout, serial or bounded concurrency
Reliability Panic recovery, publish error returns, ErrorHandler, safe handles, graceful close
Observability Global counts plus per-topic and per-handler snapshots; optional Prometheus adapter
Performance Copy-on-write subscription slices keep an unchanged synchronous publish path allocation-free

Project status

Status Milestone Scope
✅ v0.6 API convergence One options-based Subscribe; error-returning handlers and publishing
✅ v0.8 publish hot path Copy-on-write handler snapshots and zero-allocation synchronous publishing
✅ v0.9 result collection PublishCollect exposes individual synchronous delivery failures
✅ v0.11 P2 integration examples Runnable Gin service and standard-library CLI lifecycle guide
Planned v1.0 API freeze Audit preview contracts, settle any compatibility feedback, then freeze the core API and migration guidance
Planned P3 broker bridge Design one transport-specific adapter as a separate module; define delivery, retry, and shutdown semantics before code
Planned P4 mediator mode Validate a concrete application use case and publish it as an optional package, not a Bus concern
Planned P5 stateful capability Keep state, persistence, and recovery outside the core; require a separate design proposal and ownership model

The ordered path after v0.11 is API-freeze readiness, then v1.0. P3–P5 are intentionally optional modules: each needs a dedicated proposal, explicit dependency and delivery guarantees, and its own release cadence before work starts. None should add a runtime dependency to the core module.

Quality and performance

The CI matrix builds, vets, runs race tests, checks formatting, and enforces a 90% core coverage floor. The core benchmark suite measures parallel publishing; run it on your hardware before making latency claims.

go test -race ./...
(cd prometheus && go test -race ./...)
go test -bench=. -benchmem

Contributing

Issues and pull requests are welcome. Please keep the core dependency-free, add a focused test for behavioral changes, and run the quality commands above.

License

MIT

Documentation

Overview

Package bus provides a dependency-free, type-safe event bus for in-process event dispatch.

An EventBus is safe for concurrent use by multiple goroutines. Publish delivers events synchronously by default; subscriptions may opt into asynchronous delivery, priority, filtering, timeouts, and panic recovery. User-supplied handlers and implementations of optional interfaces such as Metrics and Logger must be safe for the concurrency enabled by a bus.

Example

Synchronous handlers run in the goroutine that calls Publish, so the event is fully handled by the time Publish returns. 

package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

type UserEvent struct {
	UserID string
	Action string
}

func main() {
	b := bus.NewTyped[UserEvent]()
	defer b.Close()

	handle, err := b.Subscribe("user.login", func(ctx context.Context, event UserEvent) error {
		fmt.Printf("%s performed %s\n", event.UserID, event.Action)
		return nil
	})
	if err != nil {
		fmt.Println("subscribe failed:", err)
		return
	}
	defer handle.Unsubscribe()

	b.Publish("user.login", UserEvent{UserID: "u-1", Action: "login"})

}

Output:
u-1 performed login

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// ErrBusClosed reports an operation attempted after an EventBus was closed.
	ErrBusClosed = errors.New("event bus is closed")
	// ErrNilHandler reports an attempt to subscribe a nil handler.
	ErrNilHandler = errors.New("event handler is nil")
	// ErrNilHandle reports an operation attempted on a nil subscription handle.
	ErrNilHandle = errors.New("handle is nil: the subscription was never created")
	// ErrSubscriptionInactive reports an operation attempted on an inactive subscription.
	ErrSubscriptionInactive = errors.New("subscription is inactive")
)

Functions

This section is empty.

Types

type Bus 

type Bus[T any] interface {
	BusController
	BusSubscriber[T]
	BusPublisher[T]
}

Bus englobes global (subscribe, publish, control) bus behavior

type BusController 

type BusController interface {
	HasCallback(topic string) bool
	WaitAsync()
	GetMetrics() Metrics
	SetErrorHandler(handler ErrorHandler)
	AddMiddleware(middleware EventMiddleware[any])
	SetLogger(logger Logger)
	GetLogger() Logger
	GetTopics() []string
	GetSubscriberCount(topic string) int
	Close() error
}

BusController defines bus control behavior

type BusPublisher 

type BusPublisher[T any] interface {
	Publish(topic string, event T) error
	PublishWithContext(ctx context.Context, topic string, event T) error
	PublishWithTimeout(topic string, event T, timeout time.Duration) error
}

BusPublisher defines publishing-related bus behavior

type BusResultCollector added in v0.9.0 

type BusResultCollector[T any] interface {
	PublishCollect(topic string, event T) []error
	PublishCollectWithContext(ctx context.Context, topic string, event T) []error
	PublishCollectWithTimeout(topic string, event T, timeout time.Duration) []error
}

BusResultCollector defines the optional detailed publishing behavior. It is kept separate from BusPublisher so existing publisher implementations stay source-compatible.

type BusSubscriber 

type BusSubscriber[T any] interface {
	Subscribe(topic string, fn Handler[T], options ...HandlerOption) (*Handle[T], error)
	SetDeadEventHandler(handler DeadEventHandler[T])
}

BusSubscriber defines subscription-related bus behavior

type DeadEventHandler added in v0.7.0 

type DeadEventHandler[T any] func(topic string, event T)

DeadEventHandler observes events published to a topic with no subscribed handlers, in the spirit of Guava's DeadEvent. It runs synchronously in the publishing goroutine, so it should return quickly.

type DefaultLogger 

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

DefaultLogger is the default logger implementation using Go's standard log package

func NewDefaultLogger 

func NewDefaultLogger() *DefaultLogger

NewDefaultLogger creates a new default logger instance

func NewDefaultLoggerWithOutput 

func NewDefaultLoggerWithOutput(output *os.File, prefix string) *DefaultLogger

NewDefaultLoggerWithOutput creates a new default logger with custom output

func (*DefaultLogger) Debug 

func (l *DefaultLogger) Debug(msg string, args ...interface{})

Debug logs a debug message

func (*DefaultLogger) Error 

func (l *DefaultLogger) Error(msg string, args ...interface{})

Error logs an error message

func (*DefaultLogger) GetLevel 

func (l *DefaultLogger) GetLevel() LogLevel

GetLevel returns the current log level

func (*DefaultLogger) Info 

func (l *DefaultLogger) Info(msg string, args ...interface{})

Info logs an info message

func (*DefaultLogger) SetLevel 

func (l *DefaultLogger) SetLevel(level LogLevel)

SetLevel sets the minimum log level

func (*DefaultLogger) Warn 

func (l *DefaultLogger) Warn(msg string, args ...interface{})

Warn logs a warning message

type DefaultMetrics 

type DefaultMetrics struct {
	PublishedEvents   int64
	ProcessedEvents   int64
	FailedEvents      int64
	ActiveSubscribers int32
	// contains filtered or unexported fields
}

DefaultMetrics is the default implementation of the Metrics interface.

The counter fields are updated atomically and sit on the publish hot path; read them through GetStats rather than directly. The mutex guards only the per-topic and per-handler maps.

func (*DefaultMetrics) DecrementSubscribers 

func (m *DefaultMetrics) DecrementSubscribers()

func (*DefaultMetrics) GetHandlerStats 

func (m *DefaultMetrics) GetHandlerStats() map[string]HandlerMetricsSnapshot

GetHandlerStats returns a snapshot of per-handler metrics.

func (*DefaultMetrics) GetStats 

func (m *DefaultMetrics) GetStats() (published, processed, failed int64, activeSubscribers int32)

func (*DefaultMetrics) GetTopicStats 

func (m *DefaultMetrics) GetTopicStats() map[string]TopicMetricsSnapshot

GetTopicStats returns a snapshot of per-topic metrics.

func (*DefaultMetrics) IncrementFailed 

func (m *DefaultMetrics) IncrementFailed()

func (*DefaultMetrics) IncrementProcessed 

func (m *DefaultMetrics) IncrementProcessed()

func (*DefaultMetrics) IncrementPublished 

func (m *DefaultMetrics) IncrementPublished()

func (*DefaultMetrics) IncrementSubscribers 

func (m *DefaultMetrics) IncrementSubscribers()

func (*DefaultMetrics) RecordFailed 

func (m *DefaultMetrics) RecordFailed(topic, handlerID string, duration time.Duration)

RecordFailed records a failed handler execution.

func (*DefaultMetrics) RecordProcessed 

func (m *DefaultMetrics) RecordProcessed(topic, handlerID string, duration time.Duration)

RecordProcessed records a successful handler execution.

func (*DefaultMetrics) RecordPublished 

func (m *DefaultMetrics) RecordPublished(topic string)

RecordPublished records a published event for a topic.

func (*DefaultMetrics) RemoveHandlerMetrics added in v0.11.1 

func (m *DefaultMetrics) RemoveHandlerMetrics(_ string, handlerID string)

RemoveHandlerMetrics discards per-handler metrics for an inactive subscription.

type DetailedMetrics 

type DetailedMetrics interface {
	Metrics
	RecordPublished(topic string)
	RecordProcessed(topic, handlerID string, duration time.Duration)
	RecordFailed(topic, handlerID string, duration time.Duration)
}

DetailedMetrics is an optional metrics extension for topic and handler level data.

type ErrorHandler 

type ErrorHandler func(err *EventError)

ErrorHandler defines how to handle errors during event processing

type EventBus 

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

EventBus dispatches events to topic subscribers.

EventBus is safe for concurrent use by multiple goroutines.

func New 

func New(opts ...Option[any]) *EventBus[any]

New returns new EventBus with empty handlers (for compatibility, uses any type).

func NewTyped 

func NewTyped[T any](opts ...Option[T]) *EventBus[T]

NewTyped returns new EventBus with empty handlers for the specified type.

func (*EventBus[T]) AddMiddleware 

func (bus *EventBus[T]) AddMiddleware(middleware EventMiddleware[any])

AddMiddleware adds middleware to the bus

Example

Middleware wraps the whole dispatch. Calling next runs the remaining middleware and then the handlers; not calling it intercepts the event. 

package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()

	b.AddMiddleware(func(topic string, event any, next func()) error {
		fmt.Println("before", topic)
		next()
		fmt.Println("after", topic)
		return nil
	})

	b.Subscribe("job.run", func(ctx context.Context, id string) error {
		fmt.Println("handling", id)
		return nil
	})
	b.Publish("job.run", "j-1")

}

Output:
before job.run
handling j-1
after job.run

func (*EventBus[T]) Close 

func (bus *EventBus[T]) Close() error

Close gracefully shuts down the event bus

func (*EventBus[T]) GetLogger 

func (bus *EventBus[T]) GetLogger() Logger

GetLogger returns the current logger

func (*EventBus[T]) GetMetrics 

func (bus *EventBus[T]) GetMetrics() Metrics

GetMetrics returns the current metrics

Example 
package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()

	b.Subscribe("job.run", func(ctx context.Context, id string) error { return nil })
	b.Publish("job.run", "j-1")
	b.Publish("job.run", "j-2")

	published, processed, failed, subscribers := b.GetMetrics().GetStats()
	fmt.Println(published, processed, failed, subscribers)

}

Output:
2 2 0 1

func (*EventBus[T]) GetSubscriberCount 

func (bus *EventBus[T]) GetSubscriberCount(topic string) int

GetSubscriberCount returns the number of subscribers for a topic

func (*EventBus[T]) GetTopics 

func (bus *EventBus[T]) GetTopics() []string

GetTopics returns all topics that have subscribers in lexical order.

func (*EventBus[T]) HasCallback 

func (bus *EventBus[T]) HasCallback(topic string) bool

HasCallback reports whether topic has an exact or matching-pattern subscription.

func (*EventBus[T]) Publish 

func (bus *EventBus[T]) Publish(topic string, event T) error

Publish delivers event to the topic's handlers. The returned error joins the failures of the synchronous handlers, and is safe to ignore when delivery failures do not matter to the caller. Asynchronous handler failures are reported through the ErrorHandler instead.

Example

A handler error does not stop dispatch: later handlers still run, and the publish call returns the joined failures of the synchronous handlers. 

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()
	// The default logger reports handler failures on stdout; silence it here
	// so the example output stays deterministic.
	b.SetLogger(bus.NewNoOpLogger())

	b.Subscribe("job.run", func(ctx context.Context, id string) error {
		return errors.New("disk full")
	}, bus.HandlerPriority(bus.PriorityHigh))
	b.Subscribe("job.run", func(ctx context.Context, id string) error {
		fmt.Println("second handler still runs")
		return nil
	})

	err := b.Publish("job.run", "j-1")
	fmt.Println("err:", err)

}

Output:
second handler still runs
err: disk full

func (*EventBus[T]) PublishCollect added in v0.9.0 

func (bus *EventBus[T]) PublishCollect(topic string, event T) []error

PublishCollect publishes an event and returns each synchronous dispatch failure in handler execution order. It includes errors caused by context cancellation, closing the bus, and middleware. Asynchronous handler failures remain available through ErrorHandler only, because they may occur after this method returns.

A nil result means no synchronous dispatch failure occurred. The result is independent of future publishes and may be inspected or retained by the caller.

Example

PublishCollect exposes each synchronous dispatch failure separately when a caller needs to log, retry, or classify individual handler outcomes. 

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()
	b.SetLogger(bus.NewNoOpLogger())

	b.Subscribe("job.run", func(ctx context.Context, id string) error {
		return errors.New("disk full")
	}, bus.HandlerPriority(bus.PriorityHigh))
	b.Subscribe("job.run", func(ctx context.Context, id string) error {
		return errors.New("index unavailable")
	})

	for _, err := range b.PublishCollect("job.run", "j-1") {
		fmt.Println("failure:", err)
	}

}

Output:
failure: disk full
failure: index unavailable

func (*EventBus[T]) PublishCollectWithContext added in v0.9.0 

func (bus *EventBus[T]) PublishCollectWithContext(ctx context.Context, topic string, event T) []error

PublishCollectWithContext is PublishCollect with a caller-provided context. A nil context is treated as context.Background.

func (*EventBus[T]) PublishCollectWithTimeout added in v0.9.0 

func (bus *EventBus[T]) PublishCollectWithTimeout(topic string, event T, timeout time.Duration) []error

PublishCollectWithTimeout is PublishCollect with a timeout. A timeout is reported as context.DeadlineExceeded in the returned errors.

func (*EventBus[T]) PublishWithContext 

func (bus *EventBus[T]) PublishWithContext(ctx context.Context, topic string, event T) error

PublishWithContext publishes an event with context. Canceling the context aborts dispatch to the remaining handlers and cancels the context passed to the currently running synchronous handler.

Example

Canceling the publish context aborts dispatch; a closed bus rejects the publish outright. 

package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	b.Close()

	err := b.PublishWithContext(context.Background(), "user.created", "u-1")
	fmt.Println("err:", err)

}

Output:
err: event bus is closed

func (*EventBus[T]) PublishWithTimeout 

func (bus *EventBus[T]) PublishWithTimeout(topic string, event T, timeout time.Duration) error

PublishWithTimeout publishes an event with timeout

func (*EventBus[T]) SetDeadEventHandler added in v0.7.0 

func (bus *EventBus[T]) SetDeadEventHandler(handler DeadEventHandler[T])

SetDeadEventHandler sets a handler for events published to a topic with no subscribed handlers (including pattern subscribers). Pass nil to remove it.

Example

A dead-event handler observes publishes that reached no subscriber at all — usually a misspelled topic. 

package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()

	b.SetDeadEventHandler(func(topic string, event string) {
		fmt.Printf("dead event on %q: %s\n", topic, event)
	})
	b.Subscribe("user.created", func(ctx context.Context, event string) error {
		return nil
	})

	b.Publish("user.created", "delivered") // has a subscriber: no dead event
	b.Publish("user.craeted", "u-1")       // typo: nobody subscribed

}

Output:
dead event on "user.craeted": u-1

func (*EventBus[T]) SetErrorHandler 

func (bus *EventBus[T]) SetErrorHandler(handler ErrorHandler)

SetErrorHandler sets the error handler for the bus

func (*EventBus[T]) SetLogger 

func (bus *EventBus[T]) SetLogger(logger Logger)

SetLogger sets the logger for the bus

func (*EventBus[T]) Subscribe 

func (bus *EventBus[T]) Subscribe(topic string, fn Handler[T], options ...HandlerOption) (*Handle[T], error)

Subscribe registers fn for topic and returns a handle that cancels the subscription. Behavior is configured through HandlerOption values; with no options the handler runs synchronously at PriorityNormal.

Topic may be a pattern: "*" receives every event, and a trailing ".*" receives every topic under a prefix — "user.*" matches "user.created" and "user.created.eu" but not "user" itself.

Subscribe reports why a subscription was rejected: a nil handler, a filter whose event type does not match the bus, or a closed bus. On error the returned handle is nil; a nil handle is still safe to use.

Example

Options compose: a single Subscribe call configures priority, timeout, panic policy and concurrency. 

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()

	handle, err := b.Subscribe("payment.validate",
		func(ctx context.Context, id string) error {
			fmt.Println("validating", id)
			return nil
		},
		bus.HandlerPriority(bus.PriorityHigh),
		bus.HandlerTimeout(2*time.Second),
		bus.HandlerRecoverPolicy(bus.RecoverAndStop),
		bus.HandlerSerial(),
	)
	if err != nil {
		fmt.Println("subscribe failed:", err)
		return
	}
	defer handle.Unsubscribe()

	b.Publish("payment.validate", "p-1")

}

Output:
validating p-1
Example (Hierarchical)

A trailing ".*" subscribes to every topic under a prefix: "orders.*" matches "orders.created" and "orders.created.eu", but not "orders" itself. 

package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()

	b.Subscribe("orders.*", func(ctx context.Context, payload string) error {
		fmt.Println("orders event:", payload)
		return nil
	})

	b.Publish("orders.created", "o-1")
	b.Publish("orders.created.eu", "o-2")
	b.Publish("orders", "ignored") // the prefix itself does not match
	b.Publish("invoices.created", "ignored")

}

Output:
orders event: o-1
orders event: o-2
Example (Wildcard)

The "*" topic receives every event. Wildcard handlers are merged with the topic's own handlers and ordered by priority. 

package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()

	b.Subscribe("user.created", func(ctx context.Context, payload string) error {
		fmt.Println("welcome:", payload)
		return nil
	})
	b.Subscribe("*", func(ctx context.Context, payload string) error {
		fmt.Println("audit:", payload)
		return nil
	})

	b.Publish("user.created", "u-1")

}

Output:
welcome: u-1
audit: u-1

func (*EventBus[T]) WaitAsync 

func (bus *EventBus[T]) WaitAsync()

WaitAsync waits for all async callbacks to complete

type EventError 

type EventError struct {
	Topic   string
	Event   interface{}
	Handler interface{}
	Err     error
}

EventError represents an error that occurred during event handling

func (*EventError) Error 

func (e *EventError) Error() string

type EventFilter 

type EventFilter[T any] func(topic string, event T) bool

EventFilter allows filtering events before they reach handlers

type EventMiddleware 

type EventMiddleware[T any] func(topic string, event T, next func()) error

EventMiddleware allows intercepting events before and after processing.

A middleware that wants dispatch to continue must call next before it returns. Calling next after the middleware returns is ignored.

type Handle 

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

Handle represents a subscription handle that can be used to unsubscribe

func (*Handle[T]) IsActive 

func (h *Handle[T]) IsActive() bool

IsActive returns whether this handle is still active. A nil handle is never active.

func (*Handle[T]) Unsubscribe 

func (h *Handle[T]) Unsubscribe() error

Unsubscribe removes this specific subscription.

A nil handle is reported as an error rather than a panic, so ignoring the error from Subscribe and deferring Unsubscribe stays safe.

Example (NilHandle)

Ignoring the error from Subscribe leaves a nil handle. The nil handle stays safe to use, so a deferred Unsubscribe never panics. 

package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	b.Close()

	handle, _ := b.Subscribe("user.created", func(ctx context.Context, event string) error {
		return nil
	})
	fmt.Println("active:", handle.IsActive())
	fmt.Println("err:", handle.Unsubscribe())

}

Output:
active: false
err: handle is nil: the subscription was never created

type Handler added in v0.6.0 

type Handler[T any] func(ctx context.Context, event T) error

Handler processes events delivered to a subscription.

The context reports cancellation to the handler: for synchronous handlers it is derived from the publish call and is canceled when the publish context is canceled or the handler's timeout elapses; asynchronous handlers receive the subscription context instead, because the publish call may return before they run.

A non-nil error marks the delivery as failed: it is counted in metrics, reported to the bus ErrorHandler, and - for synchronous handlers - joined into the error returned by the publish call. Returning an error does not stop dispatch to the remaining handlers.

type HandlerMetricsCleaner added in v0.11.1 

type HandlerMetricsCleaner interface {
	RemoveHandlerMetrics(topic, handlerID string)
}

HandlerMetricsCleaner is an optional Metrics extension for discarding per-handler data when a subscription becomes inactive.

Implementations should retain aggregate topic and global metrics.

type HandlerMetricsSnapshot 

type HandlerMetricsSnapshot struct {
	Topic           string
	ProcessedEvents int64
	FailedEvents    int64
	TotalDuration   time.Duration
}

HandlerMetricsSnapshot is a read-only copy of metrics for a handler.

type HandlerOption 

type HandlerOption func(*handlerOptions)

HandlerOption configures a subscription created with Subscribe.

func HandlerAsync 

func HandlerAsync(transactional bool) HandlerOption

HandlerAsync runs the handler in a goroutine. Set transactional to true to serialize this handler.

func HandlerContext 

func HandlerContext(ctx context.Context) HandlerOption

HandlerContext sets a context that can disable the handler when canceled. Asynchronous handlers also receive this context while running.

func HandlerFilter added in v0.6.0 

func HandlerFilter[T any](filter EventFilter[T]) HandlerOption

HandlerFilter runs the handler only for events accepted by the filter. The filter's event type must match the bus event type; Subscribe reports a mismatch as an error.

Example

A filter decides per event whether its handler runs at all. 

package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/townbell/bus"
)

type UserEvent struct {
	UserID string
	Action string
}

func main() {
	b := bus.NewTyped[UserEvent]()
	defer b.Close()

	b.Subscribe("user.action", func(ctx context.Context, event UserEvent) error {
		fmt.Println("admin action:", event.Action)
		return nil
	}, bus.HandlerFilter(func(topic string, event UserEvent) bool {
		return strings.HasPrefix(event.UserID, "admin-")
	}))

	b.Publish("user.action", UserEvent{UserID: "u-1", Action: "read"})
	b.Publish("user.action", UserEvent{UserID: "admin-1", Action: "delete"})

}

Output:
admin action: delete

func HandlerMaxConcurrency 

func HandlerMaxConcurrency(limit int) HandlerOption

HandlerMaxConcurrency limits concurrent executions of this handler. Values below 1 mean unlimited.

func HandlerOnce 

func HandlerOnce() HandlerOption

HandlerOnce removes the handler before its first dispatch attempt.

func HandlerPriority 

func HandlerPriority(priority Priority) HandlerOption

HandlerPriority sets the handler priority.

Example

Handlers run from the highest priority to the lowest. Handlers registered at the same priority keep their registration order. 

package main

import (
	"context"
	"fmt"

	"github.com/townbell/bus"
)

func main() {
	b := bus.NewTyped[string]()
	defer b.Close()

	say := func(line string) bus.Handler[string] {
		return func(ctx context.Context, event string) error {
			fmt.Println(line)
			return nil
		}
	}

	b.Subscribe("order.placed", say("3. analytics"), bus.HandlerPriority(bus.PriorityLow))
	b.Subscribe("order.placed", say("1. fraud check"), bus.HandlerPriority(bus.PriorityCritical))
	b.Subscribe("order.placed", say("2. fulfilment"), bus.HandlerPriority(bus.PriorityNormal))

	b.Publish("order.placed", "o-1")

}

Output:
1. fraud check
2. fulfilment
3. analytics

func HandlerRecoverPolicy 

func HandlerRecoverPolicy(policy RecoverPolicy) HandlerOption

HandlerRecoverPolicy sets how recovered panics affect the current publish call.

func HandlerSerial 

func HandlerSerial() HandlerOption

HandlerSerial limits this handler to one execution at a time.

func HandlerTimeout 

func HandlerTimeout(timeout time.Duration) HandlerOption

HandlerTimeout bounds how long a publish call waits for this handler. The handler's context is canceled when the timeout elapses.

type LogLevel 

type LogLevel int

LogLevel represents the severity level of a log message 

const (
	LogLevelDebug LogLevel = iota
	LogLevelInfo
	LogLevelWarn
	LogLevelError
)

func (LogLevel) String 

func (l LogLevel) String() string

String returns the string representation of the log level

type Logger 

type Logger interface {
	// Debug logs a debug message
	Debug(msg string, args ...interface{})
	// Info logs an info message
	Info(msg string, args ...interface{})
	// Warn logs a warning message
	Warn(msg string, args ...interface{})
	// Error logs an error message
	Error(msg string, args ...interface{})
	// SetLevel sets the minimum log level
	SetLevel(level LogLevel)
	// GetLevel returns the current log level
	GetLevel() LogLevel
}

Logger defines the interface for logging within the event bus

type Metrics 

type Metrics interface {
	IncrementPublished()
	IncrementProcessed()
	IncrementFailed()
	IncrementSubscribers()
	DecrementSubscribers()
	GetStats() (published, processed, failed int64, activeSubscribers int32)
}

Metrics defines the monitoring interface for the event bus

type NoOpLogger 

type NoOpLogger struct{}

NoOpLogger is a logger that does nothing (useful for disabling logging)

func NewNoOpLogger 

func NewNoOpLogger() *NoOpLogger

NewNoOpLogger creates a new no-op logger

func (*NoOpLogger) Debug 

func (l *NoOpLogger) Debug(msg string, args ...interface{})

Debug does nothing

func (*NoOpLogger) Error 

func (l *NoOpLogger) Error(msg string, args ...interface{})

Error does nothing

func (*NoOpLogger) GetLevel 

func (l *NoOpLogger) GetLevel() LogLevel

GetLevel returns LogLevelError (highest level to disable all logging)

func (*NoOpLogger) Info 

func (l *NoOpLogger) Info(msg string, args ...interface{})

Info does nothing

func (*NoOpLogger) SetLevel 

func (l *NoOpLogger) SetLevel(level LogLevel)

SetLevel does nothing

func (*NoOpLogger) Warn 

func (l *NoOpLogger) Warn(msg string, args ...interface{})

Warn does nothing

type Option 

type Option[T any] func(*EventBus[T])

Option defines a functional option for EventBus

func WithDeadEventHandler added in v0.7.0 

func WithDeadEventHandler[T any](handler DeadEventHandler[T]) Option[T]

WithDeadEventHandler sets a handler for events published to a topic with no subscribed handlers.

func WithErrorHandler 

func WithErrorHandler[T any](handler ErrorHandler) Option[T]

WithErrorHandler sets a custom error handler for the EventBus

func WithLogger 

func WithLogger[T any](logger Logger) Option[T]

WithLogger sets a custom logger for the EventBus

func WithMetrics 

func WithMetrics[T any](metrics Metrics) Option[T]

WithMetrics allows custom Metrics implementation

func WithMiddleware 

func WithMiddleware[T any](middleware EventMiddleware[any]) Option[T]

WithMiddleware adds a middleware to the EventBus

type PanicError added in v0.8.0 

type PanicError struct {
	// Value is the value the handler panicked with.
	Value any
}

PanicError wraps a value recovered from a panicking handler. It reaches the ErrorHandler and, for synchronous handlers, the joined publish error, so callers can tell panics apart from ordinary handler errors: 

var pe *bus.PanicError
if errors.As(err, &pe) { ... }

func (*PanicError) Error added in v0.8.0 

func (e *PanicError) Error() string

type Priority 

type Priority int

Priority defines the execution priority of handlers 

const (
	PriorityLow Priority = iota
	PriorityNormal
	PriorityHigh
	PriorityCritical
)

type RecoverPolicy 

type RecoverPolicy int

RecoverPolicy controls how handler panics are reported to the publisher. 

const (
	// RecoverAndContinue records panics and continues dispatching later handlers.
	RecoverAndContinue RecoverPolicy = iota
	// RecoverAndStop records panics and stops the current publish call.
	RecoverAndStop
)

type TopicMetricsSnapshot 

type TopicMetricsSnapshot struct {
	PublishedEvents int64
	ProcessedEvents int64
	FailedEvents    int64
	TotalDuration   time.Duration
}

TopicMetricsSnapshot is a read-only copy of metrics for a topic.

Source Files

View all Source files

Directories

Path Synopsis
prometheus module

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.