bus

package module

Go to main page

Versions in this module

v0

v0.13.0

Aug 20, 2026

Changes in this version

+ var ErrInvalidHandlerOptions = errors.New("invalid handler options")

type EventError

+ func (e *EventError) Unwrap() error

v0.12.0

Aug 20, 2026

Changes in this version

+ var ErrAsyncQueueFull = errors.New("asynchronous handler queue is full")

type HandlerOption

+ func HandlerQueueCapacity(capacity int) HandlerOption

v0.11.1

Jul 31, 2026

Changes in this version

+ var ErrBusClosed = errors.New("event bus is closed")

+ var ErrNilHandle = errors.New("handle is nil: the subscription was never created")

+ var ErrNilHandler = errors.New("event handler is nil")

+ var ErrSubscriptionInactive = errors.New("subscription is inactive")

type DefaultMetrics

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

+ type HandlerMetricsCleaner interface

+ RemoveHandlerMetrics func(topic, handlerID string)

v0.11.0

Jul 31, 2026

v0.10.0

Jul 31, 2026

v0.9.0

Jul 31, 2026

Changes in this version

+ type BusResultCollector interface

+ PublishCollect func(topic string, event T) []error

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

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

type EventBus

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

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

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

v0.8.0

Jul 26, 2026

Changes in this version

+ type PanicError struct

+ Value any

+ func (e *PanicError) Error() string

v0.7.0

Jul 26, 2026

Changes in this version

type BusSubscriber

+ SetDeadEventHandler func(handler DeadEventHandler[T])

+ type DeadEventHandler func(topic string, event T)

type EventBus

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

type Option

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

v0.6.0

Jul 26, 2026

Changes in this version

+ type Handler func(ctx context.Context, event T) error

type HandlerOption

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

v0.5.0

Jul 26, 2026

v0.4.0

Jul 26, 2026

Changes in this version

+ type Bus interface

+ type BusController interface

+ AddMiddleware func(middleware EventMiddleware[any])

+ Close func() error

+ GetLogger func() Logger

+ GetMetrics func() Metrics

+ GetSubscriberCount func(topic string) int

+ GetTopics func() []string

+ HasCallback func(topic string) bool

+ SetErrorHandler func(handler ErrorHandler)

+ SetLogger func(logger Logger)

+ WaitAsync func()

+ type BusPublisher interface

+ Publish func(topic string, event T)

+ PublishWithContext func(ctx context.Context, topic string, event T) error

+ PublishWithTimeout func(topic string, event T, timeout time.Duration) error

+ type BusSubscriber interface

+ Subscribe func(topic string, fn func(T)) error

+ SubscribeAsync func(topic string, fn func(T), transactional bool) error

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

+ SubscribeOnce func(topic string, fn func(T)) error

+ SubscribeOnceAsync func(topic string, fn func(T)) error

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

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

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

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

+ type ControlledSubscriber interface

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

+ type DefaultLogger struct

+ func NewDefaultLogger() *DefaultLogger

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

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

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

+ func (l *DefaultLogger) GetLevel() LogLevel

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

+ func (l *DefaultLogger) SetLevel(level LogLevel)

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

+ type DefaultMetrics struct

+ ActiveSubscribers int32

+ FailedEvents int64

+ ProcessedEvents int64

+ PublishedEvents int64

+ func (m *DefaultMetrics) DecrementSubscribers()

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

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

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

+ func (m *DefaultMetrics) IncrementFailed()

+ func (m *DefaultMetrics) IncrementProcessed()

+ func (m *DefaultMetrics) IncrementPublished()

+ func (m *DefaultMetrics) IncrementSubscribers()

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

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

+ func (m *DefaultMetrics) RecordPublished(topic string)

+ type DetailedMetrics interface

+ RecordFailed func(topic, handlerID string, duration time.Duration)

+ RecordProcessed func(topic, handlerID string, duration time.Duration)

+ RecordPublished func(topic string)

+ type ErrorHandler func(err *EventError)

+ type EventBus struct

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

+ type EventError struct

+ Err error

+ Event interface{}

+ Handler interface{}

+ Topic string

+ func (e *EventError) Error() string

+ type EventFilter func(topic string, event T) bool

+ type EventMiddleware func(topic string, event T, next func()) error

+ type Handle struct

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

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

+ type HandlerMetricsSnapshot struct

+ FailedEvents int64

+ ProcessedEvents int64

+ Topic string

+ TotalDuration time.Duration

+ type HandlerOption func(*handlerOptions)

+ func HandlerAsync(transactional bool) HandlerOption

+ func HandlerContext(ctx context.Context) HandlerOption

+ func HandlerMaxConcurrency(limit int) HandlerOption

+ func HandlerOnce() HandlerOption

+ func HandlerPriority(priority Priority) HandlerOption

+ func HandlerRecoverPolicy(policy RecoverPolicy) HandlerOption

+ func HandlerSerial() HandlerOption

+ func HandlerTimeout(timeout time.Duration) HandlerOption

+ type LogLevel int

+ const LogLevelDebug

+ const LogLevelError

+ const LogLevelInfo

+ const LogLevelWarn

+ func (l LogLevel) String() string

+ type Logger interface

+ Debug func(msg string, args ...interface{})

+ Error func(msg string, args ...interface{})

+ GetLevel func() LogLevel

+ Info func(msg string, args ...interface{})

+ SetLevel func(level LogLevel)

+ Warn func(msg string, args ...interface{})

+ type Metrics interface

+ DecrementSubscribers func()

+ GetStats func() (published, processed, failed int64, activeSubscribers int32)

+ IncrementFailed func()

+ IncrementProcessed func()

+ IncrementPublished func()

+ IncrementSubscribers func()

+ type NoOpLogger struct

+ func NewNoOpLogger() *NoOpLogger

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

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

+ func (l *NoOpLogger) GetLevel() LogLevel

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

+ func (l *NoOpLogger) SetLevel(level LogLevel)

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

+ type Option func(*EventBus[T])

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

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

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

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

+ type Priority int

+ const PriorityCritical

+ const PriorityHigh

+ const PriorityLow

+ const PriorityNormal

+ type RecoverPolicy int

+ const RecoverAndContinue

+ const RecoverAndStop

+ type TopicMetricsSnapshot struct

+ FailedEvents int64

+ ProcessedEvents int64

+ PublishedEvents int64

+ TotalDuration time.Duration

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL