Versions in this module Expand all Collapse all v1 v1.0.0 May 17, 2026 Changes in this version + var ErrBroadcastFull = errors.New("broadcast semaphore full") + var ErrBusClosed = errors.New("bus closed") + var ErrUnsubscribed = errors.New("subscription closed") + type AggregateStats struct + EventsDropped uint64 + EventsReceived uint64 + Subscribers int + type BroadcastResult struct + func (b BroadcastResult) Wait() error + func (b BroadcastResult) WaitContext(ctx context.Context) error + type Bus interface + Broadcast func(event T) BroadcastResult + Close func() error + CloseGracefully func(ctx context.Context) error + Len func() int + Stats func() AggregateStats + Subscribe func(options ...ChanSubscriptionOption) (Subscription[T], error) + SubscribeContext func(ctx context.Context, options ...ChanSubscriptionOption) (Subscription[T], error) + TryBroadcast func(event T) (BroadcastResult, bool) + type ChanBus struct + func NewChanBus[T any](options ...ChanBusOption) *ChanBus[T] + func (c *ChanBus[T]) Broadcast(event T) BroadcastResult + func (c *ChanBus[T]) Close() error + func (c *ChanBus[T]) CloseGracefully(ctx context.Context) error + func (c *ChanBus[T]) Len() int + func (c *ChanBus[T]) Stats() AggregateStats + func (c *ChanBus[T]) Subscribe(options ...ChanSubscriptionOption) (Subscription[T], error) + func (c *ChanBus[T]) SubscribeContext(ctx context.Context, options ...ChanSubscriptionOption) (Subscription[T], error) + func (c *ChanBus[T]) TryBroadcast(event T) (BroadcastResult, bool) + type ChanBusOption func(cfg *busConfig) + func WithBroadcastCallback(cb func(duration time.Duration, alive, dropped int)) ChanBusOption + func WithContext(ctx context.Context) ChanBusOption + func WithPoolSize(poolSize int) ChanBusOption + func WithShardCount(shardCount int) ChanBusOption + func WithSlowSubscriberPolicy(policy SlowSubscriberPolicy) ChanBusOption + func WithSlowSubscriberTimeout(timeout time.Duration) ChanBusOption + func WithSubscriptionBackpressure(backpressure int) ChanBusOption + type ChanSubscription struct + func NewChanSubscription[T any](options ...ChanSubscriptionOption) *ChanSubscription[T] + func (c *ChanSubscription[T]) Close() error + func (c *ChanSubscription[T]) Listen(ctx context.Context) (T, error) + func (c *ChanSubscription[T]) Stats() SubscriptionStats + func (c *ChanSubscription[T]) String() string + func (c *ChanSubscription[T]) Unsubscribe() + type ChanSubscriptionOption func(cfg *chanSubscriptionConfig) + func SubWithBackpressure(backpressure int) ChanSubscriptionOption + func SubWithContext(ctx context.Context) ChanSubscriptionOption + func SubWithUnsubscribeCallback(unsubscribeCallback func()) ChanSubscriptionOption + type Shards struct + func NewShards[T any](count int) *Shards[T] + func (s *Shards[T]) All() []*ChanSubscription[T] + func (s *Shards[T]) Cleanup() + func (s *Shards[T]) Len() int + func (s *Shards[T]) Snapshot() []*subList[T] + func (s *Shards[T]) Subscribe(sub *ChanSubscription[T]) int + func (s *Shards[T]) Unsubscribe(sub *ChanSubscription[T]) + func (s *Shards[T]) UnsubscribeAt(idx int, sub *ChanSubscription[T]) + type SlowSubscriberPolicy int + const SlowSubscriberPolicyBlock + const SlowSubscriberPolicyDrop + const SlowSubscriberPolicyTimeout + type Subscription interface + Close func() error + Listen func(ctx context.Context) (T, error) + Stats func() SubscriptionStats + String func() string + Unsubscribe func() + type SubscriptionStats struct + EventsDropped uint64 + EventsReceived uint64
Go to main page