bus

package module
v0.13.0 Latest Latest
Warning

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

 Go to latest Published: Aug 20, 2026 License: Apache-2.0 Imports: 12 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 guide

API preview. The v0.13 development line makes the last breaking corrections before v1: typed middleware, smaller extension interfaces, and explicit error and lifecycle contracts. See the migration guide.

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@v0.12.0

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@v0.12.0

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 joined handler and middleware errors. Later handlers still run after an ordinary error.
PublishCollect Returns synchronous handler failures in dispatch order, followed by middleware failures as the chain unwinds.
Async handlers Return before the handler finishes; failures are reported only through ErrorHandler. HandlerMaxConcurrency uses a bounded queue (64 jobs per worker by default); overflow is reported as ErrAsyncQueueFull. HandlerQueueCapacity is valid only with bounded asynchronous delivery.
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. Handler timeouts wrap context.DeadlineExceeded; handlers must honor their context to stop promptly.
Hooks Handlers, filters, middleware, error hooks, loggers, and metrics may be called concurrently and must be concurrency-safe.
Shutdown WaitAsync drains accepted async work only after publishers are quiescent. Close is the concurrent shutdown barrier; call it from the bus owner, not from one of its async handlers.

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, HandlerQueueCapacity, 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
✅ v0.12 bounded async work Bounded handler queues and measured publish-path optimizations
Release candidate v0.13 API freeze Typed middleware, narrow interfaces, explicit registration errors, lifecycle contracts, and migration guidance
Planned v1.0 API freeze Validate the v0.13 candidate in real consumers, then freeze the public API
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 is v0.13 consumer validation, 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, middleware, filters, error hooks, and implementations of optional interfaces such as Metrics and Logger must be safe for the concurrency enabled by a bus. WaitAsync requires publishers to be quiescent; use Close as the concurrent shutdown barrier.

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")
	// ErrAsyncQueueFull reports a rejected asynchronous delivery when its
	// bounded handler queue is full. It is delivered to ErrorHandler rather
	// than returned by Publish because asynchronous delivery has already
	// detached from the publisher.
	ErrAsyncQueueFull = errors.New("asynchronous handler queue is full")
	// ErrInvalidHandlerOptions reports a combination of subscription options
	// that cannot affect the configured handler.
	ErrInvalidHandlerOptions = errors.New("invalid handler options")
)

Functions

This section is empty.

Types

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)
}

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, may be called concurrently by multiple publishers, and 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 io.Writer, 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 {
	// contains filtered or unexported fields
}

DefaultMetrics is the default implementation of the Metrics interface.

Aggregate counters and detailed metrics are independently atomically updated. The detailed metric maps use sync.Map because they grow only when a topic or subscription is first observed, but are read and written on every publish. Read them through GetStats, GetTopicStats, and GetHandlerStats rather than accessing the counter fields directly.

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 handles errors during event processing. It may be called concurrently and may run inline with Publish, so it should return quickly.

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. WaitAsync is the exception: publishers must be quiescent before it is called. Use Close to reject new work and wait for accepted asynchronous work during shutdown.

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[T])

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 string, 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. It must be called by the bus owner, not from an asynchronous handler running on this bus, because Close waits for accepted asynchronous work.

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 registered under the exact key. For a pattern key such as "orders.*", it counts that pattern's subscribers; it does not count patterns that match a concrete 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 synchronous handler and middleware failures, 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 synchronous handler failures in dispatch order, followed by middleware failures as the chain unwinds. It also includes errors caused by context cancellation and closing the bus. 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, incompatible options, 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 accepted asynchronous callbacks to complete.

Publishers must be quiescent before WaitAsync is called. It is not a barrier against concurrent Publish calls starting new asynchronous work. Use Close for concurrent shutdown.

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

func (*EventError) Unwrap added in v0.13.0 

func (e *EventError) Unwrap() error

Unwrap exposes the underlying handler, middleware, or delivery error.

type EventFilter 

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

EventFilter allows filtering events before they reach handlers. A filter may be called concurrently by multiple publishers.

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. Middleware may be called concurrently by multiple publishers.

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. For an asynchronous handler it also uses a bounded work queue, preventing an overload from creating an unbounded number of goroutines. 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 HandlerQueueCapacity added in v0.12.0 

func HandlerQueueCapacity(capacity int) HandlerOption

HandlerQueueCapacity sets the maximum number of running and queued jobs for an asynchronous bounded-concurrency handler. When full, the event is rejected and reported to ErrorHandler. A positive value requires HandlerAsync with HandlerMaxConcurrency or transactional delivery; otherwise Subscribe returns ErrInvalidHandlerOptions. Values below 1 use the default of 64 jobs per worker.

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{})
	// Error logs an error message
	Error(msg string, args ...interface{})
	// 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[T]) 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.