bus

package module
v0.9.0 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 Icon

Townbell Bus

A High-Performance Event-Driven Architecture Library for Go

CI Go Version Go Reference License Go Report Card Coverage

A modern, high-performance Go event bus implementation with type safety, async processing, priority handling, filters, and enterprise-grade features.

import "github.com/townbell/bus" // package bus

中文文档

v0.6.0 replaced the subscription API. One Subscribe method with options, handlers of the form func(ctx, T) error, and Publish returns an error. Coming from v0.5.x? See MIGRATION.md — the rewrite is mechanical. This API freezes at v1.0.0; design feedback is welcome now.

✨ Features

🔧 Core Features

  • Type Safety: Go generics ensure compile-time type safety
  • Sync/Async: Support for both synchronous and asynchronous event processing
  • Handle Pattern: Precise subscription management with unsubscribe handles
  • Once Subscription: One-time event handlers that auto-remove after execution

🚀 Enterprise Features

  • Priority Processing: 4-level priority system (Critical, High, Normal, Low)
  • Event Filtering: Custom event filters for fine-grained control
  • Context Support: Context-based cancellation and timeout handling
  • Middleware: Event processing middleware chain
  • Error Handling: Comprehensive error handling and recovery
  • Monitoring: Built-in performance metrics and statistics
  • Graceful Shutdown: Proper resource cleanup and graceful termination

🔒 Reliability

  • Thread Safe: Concurrent-safe design for multi-goroutine usage
  • Panic Recovery: Automatic recovery from handler panics
  • Resource Management: Automatic cleanup and memory management

📦 Installation

go get github.com/townbell/bus

The core module has no dependencies at all — importing it pulls in nothing but the standard library. The optional Prometheus adapter lives in its own module, so client_golang only enters your build if you ask for it:

go get github.com/townbell/bus/prometheus

🚀 Quick Start

Basic Usage

package main

import (
    "context"
    "fmt"

    "github.com/townbell/bus"
)

type UserEvent struct {
    UserID string
    Action string
}

func main() {
    // Create a type-safe event bus
    eventBus := bus.NewTyped[UserEvent]()
    defer eventBus.Close()

    // Subscribe to events
    handle, err := eventBus.Subscribe("user.login", func(ctx context.Context, event UserEvent) error {
        fmt.Printf("User %s performed %s\n", event.UserID, event.Action)
        return nil
    })
    if err != nil {
        panic(err)
    }
    defer handle.Unsubscribe()

    // Publish events. The returned error joins any synchronous handler
    // failures and is safe to ignore when they do not matter to the caller.
    eventBus.Publish("user.login", UserEvent{
        UserID: "user123",
        Action: "login",
    })
}

📖 Advanced Usage

Priority Processing

Handlers with different priorities execute in priority order:

// High priority - security checks
securityHandle, _ := eventBus.Subscribe("user.action", func(ctx context.Context, event UserEvent) error {
    fmt.Println("🔒 Security check")
    return nil
}, bus.HandlerPriority(bus.PriorityCritical))

// Normal priority - business logic
businessHandle, _ := eventBus.Subscribe("user.action", func(ctx context.Context, event UserEvent) error {
    fmt.Println("📋 Business processing")
    return nil
}, bus.HandlerPriority(bus.PriorityNormal))

// Low priority - analytics
analyticsHandle, _ := eventBus.Subscribe("user.action", func(ctx context.Context, event UserEvent) error {
    fmt.Println("📊 Analytics")
    return nil
}, bus.HandlerPriority(bus.PriorityLow))

Event Filtering

Process only events that match specific criteria:

// Only process admin user events
adminHandle, _ := eventBus.Subscribe("user.action", func(ctx context.Context, event UserEvent) error {
    fmt.Printf("Admin action: %s\n", event.UserID)
    return nil
}, bus.HandlerFilter(func(topic string, event UserEvent) bool {
    return strings.HasPrefix(event.UserID, "admin_")
}))

// Only process sensitive operations
sensitiveHandle, _ := eventBus.Subscribe("user.action", func(ctx context.Context, event UserEvent) error {
    fmt.Printf("Sensitive operation alert: %s\n", event.Action)
    return nil
}, bus.HandlerFilter(func(topic string, event UserEvent) bool {
    sensitiveActions := []string{"delete", "modify_permissions"}
    for _, action := range sensitiveActions {
        if event.Action == action {
            return true
        }
    }
    return false
}))

Topic Patterns

A subscription topic can be a pattern. "*" receives every event; a trailing ".*" receives everything under a prefix — orders.* matches orders.created and orders.created.eu, but not orders itself. Pattern handlers merge with the topic's own handlers and run in priority order:

// Audit everything under orders.
eventBus.Subscribe("orders.*", func(ctx context.Context, event OrderEvent) error {
    return audit.Record(ctx, event)
}, bus.HandlerPriority(bus.PriorityLow))

Dead Events

Events published to a topic with no subscribed handlers are usually a misspelled topic. A dead-event handler makes them visible instead of silent:

eventBus.SetDeadEventHandler(func(topic string, event OrderEvent) {
    log.Printf("no subscriber for %q: %+v", topic, event)
})

Context Control

Use context for cancellation and timeout control:

// Context cancellation: canceling the subscription context disables the handler
ctx, cancel := context.WithCancel(context.Background())
handle, _ := eventBus.Subscribe("user.session", func(ctx context.Context, event UserEvent) error {
    fmt.Printf("Session event: %s\n", event.UserID)
    return nil
}, bus.HandlerContext(ctx))

// Cancel subscription
cancel()

// Timeout publishing. The handler's ctx is canceled when the deadline passes,
// so a cooperative handler can stop early instead of running to completion.
err := eventBus.PublishWithTimeout("user.action", event, 5*time.Second)
if err != nil {
    fmt.Printf("Publish timeout: %v\n", err)
}

Error Handling

Handlers report business failures by returning an error. The failure is counted in metrics, reported to the global ErrorHandler, and joined into the publish call's return value — dispatch to the remaining handlers continues:

eventBus.Subscribe("order.created", func(ctx context.Context, event OrderEvent) error {
    if err := reserveStock(ctx, event); err != nil {
        return fmt.Errorf("reserve stock: %w", err)
    }
    return nil
})

if err := eventBus.Publish("order.created", order); err != nil {
    log.Printf("delivery failures: %v", err) // errors.Is sees through the join
}

Set up a global error handler to observe every failure, including those from asynchronous handlers (whose errors never reach the publish return value):

eventBus.SetErrorHandler(func(err *bus.EventError) {
    log.Printf("Event processing error - Topic: %s, Error: %v", err.Topic, err.Err)
})

Recovered panics arrive as *bus.PanicError, so they can be told apart from ordinary handler errors:

var pe *bus.PanicError
if errors.As(err, &pe) {
    log.Printf("handler panicked with %v", pe.Value)
}

Middleware

Add processing middleware:

// Logging middleware
eventBus.AddMiddleware(func(topic string, event interface{}, next func()) error {
    start := time.Now()
    log.Printf("Processing event: %s", topic)
    
    next() // Execute handlers
    
    log.Printf("Event processed: %s, Duration: %v", topic, time.Since(start))
    return nil
})

// Rate limiting middleware
eventBus.AddMiddleware(func(topic string, event interface{}, next func()) error {
    if rateLimiter.Allow() {
        next()
        return nil
    }
    return fmt.Errorf("rate limit exceeded")
})

Monitoring Metrics

Get runtime metrics:

metrics := eventBus.GetMetrics()
published, processed, failed, subscribers := metrics.GetStats()

fmt.Printf("Published events: %d\n", published)
fmt.Printf("Processed events: %d\n", processed)
fmt.Printf("Failed events: %d\n", failed)
fmt.Printf("Active subscribers: %d\n", subscribers)

// Get topic information
topics := eventBus.GetTopics()
subscriberCount := eventBus.GetSubscriberCount("user.action")

Detailed metrics are available from the default metrics implementation:

if detailed, ok := eventBus.GetMetrics().(*bus.DefaultMetrics); ok {
    topicStats := detailed.GetTopicStats()
    handlerStats := detailed.GetHandlerStats()
    fmt.Println(topicStats["user.action"].ProcessedEvents)
    fmt.Println(handlerStats)
}

Prometheus integration is available as an optional subpackage:

import busprom "github.com/townbell/bus/prometheus"

promMetrics := busprom.New(busprom.Config{})
eventBus := bus.NewTyped[UserEvent](
    bus.WithMetrics[UserEvent](promMetrics),
)

Asynchronous Processing

// Async processing, non-transactional (concurrent execution)
_, err := eventBus.Subscribe("user.notification", func(ctx context.Context, event UserEvent) error {
    return sendEmail(ctx, event.UserID)
}, bus.HandlerAsync(false))

// Async processing, transactional (serial execution)
_, err = eventBus.Subscribe("user.audit", func(ctx context.Context, event UserEvent) error {
    return writeAuditLog(ctx, event)
}, bus.HandlerAsync(true))

Handler Execution Control

Every subscription concern is a HandlerOption, and they compose freely in one Subscribe call:

handle, err := eventBus.Subscribe("payment.validate",
    func(ctx context.Context, event PaymentEvent) error {
        return validatePayment(ctx, event)
    },
    bus.HandlerTimeout(2*time.Second),           // cancels ctx when it elapses
    bus.HandlerRecoverPolicy(bus.RecoverAndStop), // a panic aborts the publish
    bus.HandlerSerial(),                          // one execution at a time
    bus.HandlerPriority(bus.PriorityHigh),
)

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

Collecting Handler Errors

Use PublishCollect when each synchronous handler failure needs individual handling rather than a single joined error. Async failures still go to the configured ErrorHandler:

for _, err := range eventBus.PublishCollect("payment.validate", payment) {
    log.Printf("validation handler failed: %v", err)
}

✅ Behavior Contract

  • Publish and PublishWithContext return the joined failures of the synchronous handlers (errors.Is sees through the join). The error is safe to ignore. Asynchronous handler failures are reported through ErrorHandler only, because the publish call may return before they run.
  • PublishCollect, PublishCollectWithContext, and PublishCollectWithTimeout return each synchronous dispatch failure in handler execution order. They include context, close, and middleware failures; asynchronous handler failures remain available through ErrorHandler only.
  • A handler error does not stop dispatch: the remaining handlers still run. Dispatch stops early only when the publish context is canceled, the bus closes, or a handler panics under RecoverAndStop.
  • Synchronous handlers run in the goroutine that calls publish and receive a context derived from the publish call; asynchronous handlers run in separate goroutines and receive the subscription context (HandlerContext), and HandlerAsync(true) serializes calls to the same handler.
  • HandlerTimeout bounds how long the publish call waits and cancels the handler's context when it elapses; a handler that ignores its context keeps running in the background and is still awaited by WaitAsync and Close.
  • Handler panics are recovered as *PanicError, counted as failures, and reported through ErrorHandler; RecoverAndContinue (the default) keeps dispatching, RecoverAndStop aborts the publish call. Either way the recovered panic appears in the publish error and is detectable with errors.As.
  • Middleware must call next() to continue to the next middleware and handlers; skipping next() intercepts the event.
  • HandlerOnce handlers execute successfully at most once, including when multiple one-time handlers share a topic.
  • After Close, the bus rejects new publish and subscribe calls; already-started async handlers are allowed to finish.
  • Subscribe returns (nil, error) when the subscription is rejected (nil callback, mismatched filter type, or a closed bus). A nil handle is safe to use: Unsubscribe returns an error and IsActive returns false, so ignoring the error and deferring Unsubscribe never panics.
  • Pattern subscriptions: "*" matches every topic, "prefix.*" matches every topic strictly under prefix. (not prefix itself). Matched pattern handlers merge with exact-topic handlers and run in priority order; pattern names are sorted so same-priority ordering is deterministic. HasCallback and GetSubscriberCount remain exact-key lookups.
  • The dead-event handler fires when a publish finds zero subscribed handlers, before middleware runs. A subscriber whose filter rejects the event still counts as a subscriber, so no dead event fires. It runs synchronously in the publishing goroutine.

🗺️ RoadMap

Townbell will keep its focus on being an in-process, type-safe, lightweight event bus. Future work may borrow ideas from Watermill, Blinker, MediatR, and Guava EventBus, but the core package will not try to become a full distributed messaging system.

Priority Status Area Notes
P0 Done Core correctness Publish no longer holds the bus lock while running handlers; SubscribeOnce removal, middleware chaining, and race tests are covered
P0 Done API contract Error returns for Publish / PublishWithContext, panic recovery, sync/async execution, and closed-bus behavior are documented
P1 Done Documentation alignment README content now documents the current P1 behavior and examples
P1 Done Observability Per-topic and per-handler published, processed, failed, and duration metrics are available, with an optional Prometheus adapter
P1 Done Execution control SubscribeWithOptions supports handler-level timeout, recover policy, async/serial execution, and max concurrency
P1 Done Continuous integration GitHub Actions runs build, vet, race tests, a gofmt gate and a coverage floor across a Go version matrix, and vets example/ explicitly
P1 Done Dependency-free core The Prometheus adapter moved into its own module, so importing bus pulls in nothing but the standard library
P1 Done Runnable documentation Godoc Example functions execute in CI with verified output and render on pkg.go.dev
P0 Preview (v0.6.0) Subscription API convergence One Subscribe(topic, fn, opts...) (*Handle[T], error) replaced the ten previous variants. Freezes at v1.0.0
P0 Preview (v0.6.0) Handler error reporting Handlers are func(ctx, T) error: business failures reach metrics, the ErrorHandler, and the publish return value without panicking. Unblocks result collection. Freezes at v1.0.0
P0 Preview (v0.6.0) Publish error semantics Publish returns the joined synchronous-handler failures; ignoring it stays legal. Freezes at v1.0.0
P2 Partial (v0.9.0) Result collection PublishCollect returns each synchronous handler error in dispatch order; collecting arbitrary handler values needs a separate v1 design
P2 Done Topic enhancements Wildcard (*) topics, hierarchical prefix.* patterns, and a dead-event hook for publishes that reach no subscriber
P2 Partial Integration examples net/http example shipped; Gin, CLI apps, and workers remain
P3 Planned Broker bridges Borrow from Watermill and explore NATS / Kafka / RabbitMQ adapters, preferably in separate subpackages
P3 Planned Mediator mode Borrow from MediatR and add request / response, command, query, and notification support only if needed
P4 Planned Stateful features Evaluate sticky events, event replay, and local persistence only when there is a clear use case

🏗️ Architecture Design

The library is organized into separate modules for better maintainability:

File Structure

  • types.go - Core type definitions (Priority, EventError, filters, middleware)
  • interfaces.go - Interface definitions (BusSubscriber, BusPublisher, BusController, Bus)
  • metrics.go - Monitoring and metrics functionality
  • handle.go - Subscription handle management and internal handler structures
  • bus.go - Core EventBus implementation

Interface Separation

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

// Publisher interface
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
}

// Optional detailed publisher interface
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
}

// Controller interface
type BusController interface {
    GetMetrics() Metrics
    SetErrorHandler(handler ErrorHandler)
    AddMiddleware(middleware EventMiddleware[any])
    Close() error
    // ...
}

Type System

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

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

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

// Error handler
type ErrorHandler func(err *EventError)

// Priority levels
type Priority int
const (
    PriorityLow Priority = iota
    PriorityNormal
    PriorityHigh
    PriorityCritical
)

🔧 Best Practices

1. Event Design Patterns

Following industry best practices, supports these event design patterns:

Event Notification
type UserCreatedEvent struct {
    UserID    string    `json:"user_id"`
    Timestamp time.Time `json:"timestamp"`
    // Minimal data, subscribers fetch details themselves
}
Event-Carried State Transfer
type UserUpdatedEvent struct {
    UserID       string                 `json:"user_id"`
    Timestamp    time.Time              `json:"timestamp"`
    OldState     map[string]interface{} `json:"old_state"`
    NewState     map[string]interface{} `json:"new_state"`
    ChangedFields []string              `json:"changed_fields"`
}

2. Naming Conventions

// Use dot-separated hierarchical naming
"user.created"
"user.updated"
"user.deleted"
"order.placed"
"order.cancelled"
"payment.processed"
"payment.failed"

// Or use namespaces
"ecommerce.order.created"
"auth.user.login"
"notification.email.sent"

3. Error Handling Strategy

// Set up retry mechanism
eventBus.SetErrorHandler(func(err *EventError) {
    switch err.Err.(type) {
    case *TemporaryError:
        // Temporary error, retry later
        retryQueue.Add(err.Topic, err.Event)
    case *PermanentError:
        // Permanent error, log and alert
        logger.Error("Permanent error", err)
        alerting.Send(err)
    default:
        // Unknown error, log details
        logger.Warn("Unknown error", err)
    }
})

4. Performance Optimization

// Use async for non-critical paths
eventBus.Subscribe("analytics.track", func(ctx context.Context, event UserEvent) error {
    return analytics.Track(ctx, event) // Non-critical analytics
}, bus.HandlerAsync(false))

// Use sync for critical paths
eventBus.Subscribe("payment.validate", func(ctx context.Context, event PaymentEvent) error {
    return validatePayment(ctx, event) // Critical payment validation
})

// Use filters to reduce unnecessary processing
eventBus.Subscribe("user.activity", handler, bus.HandlerFilter(
    func(topic string, event UserEvent) bool {
        return event.IsImportant() // Only process important events
    }))

🔍 Comparison with Other Libraries

Feature Townbell Guava EventBus RxJava Node.js EventEmitter
Type Safety ✅ Generics ✅ ✅ ❌
Async Processing ✅ ❌ ✅ ✅
Priority ✅ ❌ ❌ ❌
Filters ✅ ❌ ✅ ❌
Middleware ✅ ❌ ✅ ❌
Error Handling ✅ ⚠️ ✅ ⚠️
Monitoring ✅ ❌ ❌ ❌
Context Support ✅ ❌ ❌ ❌

🧪 Testing

Run the complete test suite. The Prometheus adapter is a separate module, so it needs its own invocation:

go test -race ./...
(cd prometheus && go test -race ./...)

Files under example/ carry //go:build ignore, which means go vet ./... skips them. Vet them by name:

for f in example/*.go; do go vet "$f"; done

Generate a coverage report:

go test -coverprofile=coverage.out ./...
go tool cover -html=coverage.out -o coverage.html

Current coverage: 95.5% for the core module, 79.7% for the Prometheus adapter. CI enforces a 90% floor on the core module, so this figure cannot silently drift.

Run the benchmarks:

go test -bench=. -benchmem

📈 Performance

Measured on an Apple M5 (10 cores), Go 1.22.12, darwin/arm64, on 2026-07-27 (v0.8.0). Every benchmark uses b.RunParallel, so ns/op is the aggregate cost across all cores rather than single-goroutine latency.

Benchmark ns/op B/op allocs/op
SyncPublish (1 subscriber) 350 0 0
AsyncPublish (1 subscriber) 784 128 1
MultipleSubscribers (10 subscribers) 2580 0 0
WithPriority 254 0 0
WithFilter 63 0 0
ConcurrentSubscribeUnsubscribe 2964 713 14
ChannelBaseline (raw Go channel) 49 0 0

Synchronous publishing allocates nothing. Handler lists are copy-on-write: subscription changes build fresh slices, so a publish uses the current list directly instead of copying it, and the middleware machinery and debug logging are skipped entirely when unused. v0.8.0 cut a sync publish from 839 ns and 11 allocations to 350 ns and zero.

Two more things worth reading off this table:

  • Asynchronous publishing is slower than synchronous publishing, not faster. Every async dispatch starts a goroutine and touches a WaitGroup and a mutex. Reach for async to keep a slow handler off the publishing goroutine, not to raise throughput.
  • A raw channel is still cheaper — about 2x on a single goroutine. The bus buys you fan-out, priorities, filters, middleware and metrics. If all you need is to hand a value to one known goroutine, a channel is the better tool.

Numbers on your hardware will differ. Re-run the benchmarks rather than trusting this table.

🤝 Contributing

We welcome Issues and Pull Requests!

  1. Fork the repository
  2. Create your feature branch (git checkout -b feature/amazing-feature)
  3. Commit your changes (git commit -m 'Add amazing feature')
  4. Push to the branch (git push origin feature/amazing-feature)
  5. Open a Pull Request

📄 License

MIT License - see the LICENSE file for details

🙏 Acknowledgments

This project draws inspiration from these excellent open source projects and design patterns:

Documentation

Overview

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

This section is empty.

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.

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 - box for handlers and callbacks.

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

func (*EventBus[T]) HasCallback 

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

HasCallback returns true if exists any callback subscribed to the topic.

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

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 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.