bus

package module
v0.5.0 Latest Latest
Warning

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

 Go to latest Published: Jul 26, 2026 License: Apache-2.0 Imports: 8 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

中文文档

✨ 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 (
    "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 := eventBus.SubscribeWithHandle("user.login", func(event UserEvent) {
        fmt.Printf("User %s performed %s\n", event.UserID, event.Action)
    })
    defer handle.Unsubscribe()

    // Publish events
    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.SubscribeWithPriority("user.action", func(event UserEvent) {
    fmt.Println("🔒 Security check")
}, bus.PriorityCritical)

// Normal priority - business logic
businessHandle := eventBus.SubscribeWithPriority("user.action", func(event UserEvent) {
    fmt.Println("📋 Business processing")
}, bus.PriorityNormal)

// Low priority - analytics
analyticsHandle := eventBus.SubscribeWithPriority("user.action", func(event UserEvent) {
    fmt.Println("📊 Analytics")
}, bus.PriorityLow)

Event Filtering

Process only events that match specific criteria:

// Only process admin user events
adminHandle := eventBus.SubscribeWithFilter("user.action", func(event UserEvent) {
    fmt.Printf("Admin action: %s\n", event.UserID)
}, func(topic string, event UserEvent) bool {
    return strings.HasPrefix(event.UserID, "admin_")
})

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

Context Control

Use context for cancellation and timeout control:

// Context cancellation
ctx, cancel := context.WithCancel(context.Background())
handle := eventBus.SubscribeWithContext(ctx, "user.session", func(event UserEvent) {
    fmt.Printf("Session event: %s\n", event.UserID)
})

// Cancel subscription
cancel()

// Timeout publishing
err := eventBus.PublishWithTimeout("user.action", event, 5*time.Second)
if err != nil {
    fmt.Printf("Publish timeout: %v\n", err)
}

Error Handling

Set up global error handler:

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

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.SubscribeAsync("user.notification", func(event UserEvent) {
    sendEmail(event.UserID)
}, false)

// Async processing, transactional (serial execution)
err := eventBus.SubscribeAsync("user.audit", func(event UserEvent) {
    writeAuditLog(event)
}, true)

Handler Execution Control

SubscribeWithOptions adds handler-level controls without changing the existing simple APIs:

handle, err := eventBus.SubscribeWithOptions("payment.validate", func(event PaymentEvent) {
    validatePayment(event)
},
    bus.HandlerTimeout(2*time.Second),
    bus.HandlerRecoverPolicy(bus.RecoverAndStop),
    bus.HandlerSerial(),
    bus.HandlerPriority(bus.PriorityHigh),
)

✅ Behavior Contract

  • Publish ignores returned errors; use PublishWithContext or PublishWithTimeout when cancellation, timeout, or closed-bus errors matter.
  • Synchronous handlers run in the current goroutine by default; asynchronous handlers run in separate goroutines, and transactional=true serializes calls to the same handler.
  • Handler timeouts bound how long the publish call waits; they do not forcibly stop a handler that has already started running.
  • Handler panics are recovered by default, failed metrics are incremented, and the error is reported through ErrorHandler; RecoverAndStop makes a recovered panic stop the current publish call.
  • Middleware must call next() to continue to the next middleware and handlers; skipping next() intercepts the event.
  • SubscribeOnce / SubscribeOnceAsync 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 helpers that return only a *Handle[T] yield nil when the subscription is rejected (nil callback, or a closed bus). A nil handle is safe to use: Unsubscribe returns an error and IsActive returns false, so a deferred Unsubscribe never panics. Use SubscribeWithOptions when you want the rejection reason.

🗺️ 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 Planned Subscription API convergence Three different shapes coexist today: error only, *Handle[T] only, and (*Handle[T], error). Converge on the last one. Breaking, so it gates v1.0.0
P0 Planned Handler error reporting Handlers are func(T) and cannot report a business failure except by panicking, so ErrorHandler only ever sees panics and timeouts. Breaking, gates v1.0.0, and unblocks result collection
P0 Planned Publish error semantics Decide whether Publish returns an error or whether discarding it becomes a permanent contract. Gates v1.0.0
P2 Planned Result collection Borrow from Blinker and add a PublishCollect-style API for collecting handler results or errors
P2 Partial Topic enhancements Wildcard (*) topics are implemented; hierarchical topics and no-subscriber hooks are still planned
P2 Planned Integration examples Add practical examples for net/http, Gin, CLI apps, and workers
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 func(T)) error
    SubscribeWithPriority(topic string, fn func(T), priority Priority) *Handle[T]
    SubscribeWithFilter(topic string, fn func(T), filter EventFilter[T]) *Handle[T]
    SubscribeWithContext(ctx context.Context, topic string, fn func(T)) *Handle[T]
    // ...
}

// Publisher interface
type BusPublisher[T any] interface {
    Publish(topic string, event T)
    PublishWithContext(ctx context.Context, topic string, event T) error
    PublishWithTimeout(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 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.SubscribeAsync("analytics.track", func(event UserEvent) {
    // Non-critical analytics
    analytics.Track(event)
}, false)

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

// Use filters to reduce unnecessary processing
eventBus.SubscribeWithFilter("user.activity", handler, 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: 93.3% 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-26. 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) 839 392 11
AsyncPublish (1 subscriber) 1320 520 12
MultipleSubscribers 4470 752 29
WithPriority 532 472 15
WithFilter 240 376 10
ConcurrentSubscribeUnsubscribe 3613 1207 38
ChannelBaseline (raw Go channel) 50 0 0

Two things are 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 roughly 17x cheaper. 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 (
	"fmt"

	"github.com/townbell/bus"
)

type UserEvent struct {
	UserID string
	Action string
}

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

	handle := b.SubscribeWithHandle("user.login", func(event UserEvent) {
		fmt.Printf("%s performed %s\n", event.UserID, event.Action)
	})
	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)
	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 BusSubscriber 

type BusSubscriber[T any] interface {
	Subscribe(topic string, fn func(T)) error
	SubscribeAsync(topic string, fn func(T), transactional bool) error
	SubscribeOnce(topic string, fn func(T)) error
	SubscribeOnceAsync(topic string, fn func(T)) error
	SubscribeWithHandle(topic string, fn func(T)) *Handle[T]
	SubscribeAsyncWithHandle(topic string, fn func(T), transactional bool) *Handle[T]
	SubscribeWithPriority(topic string, fn func(T), priority Priority) *Handle[T]
	SubscribeWithFilter(topic string, fn func(T), filter EventFilter[T]) *Handle[T]
	SubscribeWithContext(ctx context.Context, topic string, fn func(T)) *Handle[T]
}

BusSubscriber defines subscription-related bus behavior

type ControlledSubscriber 

type ControlledSubscriber[T any] interface {
	SubscribeWithOptions(topic string, fn func(T), options ...HandlerOption) (*Handle[T], error)
}

ControlledSubscriber defines optional handler-level execution control behavior.

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

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 (
	"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(id string) {
		fmt.Println("handling", id)
	})
	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 (
	"fmt"

	"github.com/townbell/bus"
)

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

	_ = b.Subscribe("job.run", func(string) {})
	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)

Publish executes callback defined for a topic.

func (*EventBus[T]) PublishWithContext 

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

PublishWithContext publishes an event with context

Example

Publish discards errors. Use PublishWithContext when cancellation, timeout or closed-bus errors matter. 

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]) 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 func(T)) error

Subscribe subscribes to a topic.

Example (Wildcard)

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

package main

import (
	"fmt"

	"github.com/townbell/bus"
)

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

	_ = b.Subscribe("user.created", func(payload string) {
		fmt.Println("welcome:", payload)
	})
	_ = b.Subscribe("*", func(payload string) {
		fmt.Println("audit:", payload)
	})

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

}

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

func (*EventBus[T]) SubscribeAsync 

func (bus *EventBus[T]) SubscribeAsync(topic string, fn func(T), transactional bool) error

SubscribeAsync subscribes to a topic with an asynchronous callback

func (*EventBus[T]) SubscribeAsyncWithHandle 

func (bus *EventBus[T]) SubscribeAsyncWithHandle(topic string, fn func(T), transactional bool) *Handle[T]

SubscribeAsyncWithHandle subscribes to a topic with an asynchronous callback and returns a handle.

func (*EventBus[T]) SubscribeOnce 

func (bus *EventBus[T]) SubscribeOnce(topic string, fn func(T)) error

SubscribeOnce subscribes to a topic once. Handler will be removed after executing.

func (*EventBus[T]) SubscribeOnceAsync 

func (bus *EventBus[T]) SubscribeOnceAsync(topic string, fn func(T)) error

SubscribeOnceAsync subscribes to a topic once with an asynchronous callback

func (*EventBus[T]) SubscribeWithContext 

func (bus *EventBus[T]) SubscribeWithContext(ctx context.Context, topic string, fn func(T)) *Handle[T]

SubscribeWithContext subscribes to a topic with context for cancellation

func (*EventBus[T]) SubscribeWithFilter 

func (bus *EventBus[T]) SubscribeWithFilter(topic string, fn func(T), filter EventFilter[T]) *Handle[T]

SubscribeWithFilter subscribes to a topic with an event filter

Example

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

package main

import (
	"fmt"
	"strings"

	"github.com/townbell/bus"
)

type UserEvent struct {
	UserID string
	Action string
}

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

	b.SubscribeWithFilter("user.action", func(event UserEvent) {
		fmt.Println("admin action:", event.Action)
	}, 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 (*EventBus[T]) SubscribeWithHandle 

func (bus *EventBus[T]) SubscribeWithHandle(topic string, fn func(T)) *Handle[T]

SubscribeWithHandle subscribes to a topic and returns a handle for unsubscription.

func (*EventBus[T]) SubscribeWithOptions 

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

SubscribeWithOptions subscribes to a topic with handler-level execution controls.

Example

SubscribeWithOptions is the extensible subscription path. It is the only helper that reports why a subscription was rejected. 

package main

import (
	"fmt"
	"time"

	"github.com/townbell/bus"
)

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

	handle, err := b.SubscribeWithOptions("payment.validate",
		func(id string) { fmt.Println("validating", id) },
		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

func (*EventBus[T]) SubscribeWithPriority 

func (bus *EventBus[T]) SubscribeWithPriority(topic string, fn func(T), priority Priority) *Handle[T]

SubscribeWithPriority subscribes to a topic with specified priority

Example

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

package main

import (
	"fmt"

	"github.com/townbell/bus"
)

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

	b.SubscribeWithPriority("order.placed", func(string) {
		fmt.Println("3. analytics")
	}, bus.PriorityLow)
	b.SubscribeWithPriority("order.placed", func(string) {
		fmt.Println("1. fraud check")
	}, bus.PriorityCritical)
	b.SubscribeWithPriority("order.placed", func(string) {
		fmt.Println("2. fulfilment")
	}, bus.PriorityNormal)

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

}

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

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. Subscribe helpers that return only a handle yield nil when the subscription is rejected (nil callback, or a closed bus), so a deferred Unsubscribe must stay safe.

Example (NilHandle)

Subscribe helpers that return only a handle yield nil when the subscription is rejected. The returned handle stays safe to use, so a deferred Unsubscribe never panics. 

package main

import (
	"fmt"

	"github.com/townbell/bus"
)

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

	handle := b.SubscribeWithHandle("user.created", func(string) {})
	fmt.Println("active:", handle.IsActive())
	fmt.Println("err:", handle.Unsubscribe())

}

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

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

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.

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.

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 PublishWithContext waits for this handler.

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