eventbroker

Dec 30, 2025

Changes in this version

+ func MakeShutdownCallback(broker Broker) func(context.Context) error

+ type DatabaseProvider struct

+ func NewDatabaseProvider(cfg DatabaseProviderConfig) (*DatabaseProvider, error)

+ func (dp *DatabaseProvider) Close() error

+ func (dp *DatabaseProvider) Delete(ctx context.Context, id string) error

+ func (dp *DatabaseProvider) Get(ctx context.Context, id string) (*Event, error)

+ func (dp *DatabaseProvider) List(ctx context.Context, filter *EventFilter) ([]*Event, error)

+ func (dp *DatabaseProvider) Publish(ctx context.Context, event *Event) error

+ func (dp *DatabaseProvider) Stats(ctx context.Context) (*ProviderStats, error)

+ func (dp *DatabaseProvider) Store(ctx context.Context, event *Event) error

+ func (dp *DatabaseProvider) Stream(ctx context.Context, pattern string) (<-chan *Event, error)

+ func (dp *DatabaseProvider) UpdateStatus(ctx context.Context, id string, status EventStatus, errorMsg string) error

+ type DatabaseProviderConfig struct

+ Channel string

+ DB common.Database

+ InstanceID string

+ PollInterval time.Duration

+ TableName string

+ UseNotify bool

+ type DatabaseProviderStats struct

+ ActiveSubscribers atomic.Int32

+ EventsConsumed atomic.Int64

+ EventsPublished atomic.Int64

+ PollErrors atomic.Int64

+ TotalEvents atomic.Int64

+ type NATSProvider struct

+ func NewNATSProvider(cfg NATSProviderConfig) (*NATSProvider, error)

+ func (np *NATSProvider) Close() error

+ func (np *NATSProvider) Delete(ctx context.Context, id string) error

+ func (np *NATSProvider) Get(ctx context.Context, id string) (*Event, error)

+ func (np *NATSProvider) List(ctx context.Context, filter *EventFilter) ([]*Event, error)

+ func (np *NATSProvider) Publish(ctx context.Context, event *Event) error

+ func (np *NATSProvider) Stats(ctx context.Context) (*ProviderStats, error)

+ func (np *NATSProvider) Store(ctx context.Context, event *Event) error

+ func (np *NATSProvider) Stream(ctx context.Context, pattern string) (<-chan *Event, error)

+ func (np *NATSProvider) UpdateStatus(ctx context.Context, id string, status EventStatus, errorMsg string) error

+ type NATSProviderConfig struct

+ InstanceID string

+ MaxAge time.Duration

+ Storage string

+ StreamName string

+ SubjectPrefix string

+ URL string

+ type NATSProviderStats struct

+ ActiveSubscribers atomic.Int32

+ ConsumerErrors atomic.Int64

+ EventsConsumed atomic.Int64

+ EventsPublished atomic.Int64

+ TotalEvents atomic.Int64

+ type RedisProvider struct

+ func NewRedisProvider(cfg RedisProviderConfig) (*RedisProvider, error)

+ func (rp *RedisProvider) Close() error

+ func (rp *RedisProvider) Delete(ctx context.Context, id string) error

+ func (rp *RedisProvider) Get(ctx context.Context, id string) (*Event, error)

+ func (rp *RedisProvider) List(ctx context.Context, filter *EventFilter) ([]*Event, error)

+ func (rp *RedisProvider) Publish(ctx context.Context, event *Event) error

+ func (rp *RedisProvider) Stats(ctx context.Context) (*ProviderStats, error)

+ func (rp *RedisProvider) Store(ctx context.Context, event *Event) error

+ func (rp *RedisProvider) Stream(ctx context.Context, pattern string) (<-chan *Event, error)

+ func (rp *RedisProvider) UpdateStatus(ctx context.Context, id string, status EventStatus, errorMsg string) error

+ type RedisProviderConfig struct

+ ConsumerGroup string

+ ConsumerName string

+ DB int

+ Host string

+ InstanceID string

+ MaxLen int64

+ Password string

+ Port int

+ StreamName string

+ type RedisProviderStats struct

+ ActiveSubscribers atomic.Int32

+ ConsumerErrors atomic.Int64

+ EventsConsumed atomic.Int64

+ EventsPublished atomic.Int64

+ TotalEvents atomic.Int64

Dec 12, 2025

Changes in this version

+ const ExampleYAMLConfiguration

+ var ErrQueueFull = &BrokerError

+ var ErrWorkerPoolStopped = &BrokerError

+ func EventType(schema, entity, operation string) string

+ func Example()

+ func ExampleConfiguration()

+ func ExampleErrorHandling()

+ func ExampleSubscriptionPatterns()

+ func ExampleWithHooks()

+ func Initialize(cfg config.EventBrokerConfig) error

+ func IsInitialized() bool

+ func Publish(ctx context.Context, event *Event) error

+ func PublishAsync(ctx context.Context, event *Event) error

+ func PublishSync(ctx context.Context, event *Event) error

+ func RegisterCRUDHooks(broker Broker, hookRegistry *restheadspec.HookRegistry, config *CRUDHookConfig) error

+ func RegisterShutdown(broker Broker)

+ func SetDefaultBroker(broker Broker)

+ func Unsubscribe(id SubscriptionID) error

+ func WithBufferSize(size int) func(*Options)

+ func WithInstanceID(id string) func(*Options)

+ func WithMode(m ProcessingMode) func(*Options)

+ func WithProvider(p Provider) func(*Options)

+ func WithRetryPolicy(policy *RetryPolicy) func(*Options)

+ func WithWorkerCount(count int) func(*Options)

+ type Broker interface

+ InstanceID func() string

+ Publish func(ctx context.Context, event *Event) error

+ PublishAsync func(ctx context.Context, event *Event) error

+ PublishSync func(ctx context.Context, event *Event) error

+ Start func(ctx context.Context) error

+ Stats func(ctx context.Context) (*BrokerStats, error)

+ Stop func(ctx context.Context) error

+ Subscribe func(pattern string, handler EventHandler) (SubscriptionID, error)

+ Unsubscribe func(id SubscriptionID) error

+ func GetDefaultBroker() Broker

+ type BrokerError struct

+ Code string

+ Message string

+ func (e *BrokerError) Error() string

+ type BrokerStats struct

+ ActiveSubscribers int

+ ActiveWorkers int

+ AdditionalStats map[string]interface{}

+ InstanceID string

+ IsRunning bool

+ Mode ProcessingMode

+ ProviderStats *ProviderStats

+ QueueSize int

+ TotalFailed int64

+ TotalProcessed int64

+ TotalPublished int64

+ func Stats(ctx context.Context) (*BrokerStats, error)

+ type CRUDHookConfig struct

+ EnableCreate bool

+ EnableDelete bool

+ EnableRead bool

+ EnableUpdate bool

+ func DefaultCRUDHookConfig() *CRUDHookConfig

+ type Event struct

+ CompletedAt *time.Time

+ CreatedAt time.Time

+ Entity string

+ Error string

+ ID string

+ InstanceID string

+ Metadata map[string]interface{}

+ Operation string

+ Payload json.RawMessage

+ ProcessedAt *time.Time

+ RetryCount int

+ Schema string

+ SessionID string

+ Source EventSource

+ Status EventStatus

+ Type string

+ UserID int

+ func NewEvent(source EventSource, eventType string) *Event

+ func (e *Event) Clone() *Event

+ func (e *Event) GetPayload(v interface{}) error

+ func (e *Event) IncrementRetry()

+ func (e *Event) MarkCompleted()

+ func (e *Event) MarkFailed(err error)

+ func (e *Event) MarkProcessing()

+ func (e *Event) SetPayload(v interface{}) error

+ func (e *Event) Validate() error

+ type EventBroker struct

+ func NewBroker(opts Options) (*EventBroker, error)

+ func (b *EventBroker) InstanceID() string

+ func (b *EventBroker) Publish(ctx context.Context, event *Event) error

+ func (b *EventBroker) PublishAsync(ctx context.Context, event *Event) error

+ func (b *EventBroker) PublishSync(ctx context.Context, event *Event) error

+ func (b *EventBroker) Start(ctx context.Context) error

+ func (b *EventBroker) Stats(ctx context.Context) (*BrokerStats, error)

+ func (b *EventBroker) Stop(ctx context.Context) error

+ func (b *EventBroker) Subscribe(pattern string, handler EventHandler) (SubscriptionID, error)

+ func (b *EventBroker) Unsubscribe(id SubscriptionID) error

+ type EventFilter struct

+ EndTime *time.Time

+ Entity string

+ InstanceID string

+ Limit int

+ Offset int

+ Operation string

+ Schema string

+ Source *EventSource

+ StartTime *time.Time

+ Status *EventStatus

+ UserID *int

+ type EventHandler interface

+ Handle func(ctx context.Context, event *Event) error

+ type EventHandlerFunc func(ctx context.Context, event *Event) error

+ func (f EventHandlerFunc) Handle(ctx context.Context, event *Event) error

+ type EventSource string

+ const EventSourceDatabase

+ const EventSourceFrontend

+ const EventSourceInternal

+ const EventSourceSystem

+ const EventSourceWebSocket

+ type EventStatus string

+ const EventStatusCompleted

+ const EventStatusFailed

+ const EventStatusPending

+ const EventStatusProcessing

+ type MemoryProvider struct

+ func NewMemoryProvider(opts MemoryProviderOptions) *MemoryProvider

+ func (mp *MemoryProvider) Close() error

+ func (mp *MemoryProvider) Delete(ctx context.Context, id string) error

+ func (mp *MemoryProvider) Get(ctx context.Context, id string) (*Event, error)

+ func (mp *MemoryProvider) List(ctx context.Context, filter *EventFilter) ([]*Event, error)

+ func (mp *MemoryProvider) Publish(ctx context.Context, event *Event) error

+ func (mp *MemoryProvider) Stats(ctx context.Context) (*ProviderStats, error)

+ func (mp *MemoryProvider) Store(ctx context.Context, event *Event) error

+ func (mp *MemoryProvider) Stream(ctx context.Context, pattern string) (<-chan *Event, error)

+ func (mp *MemoryProvider) UpdateStatus(ctx context.Context, id string, status EventStatus, errorMsg string) error

+ type MemoryProviderOptions struct

+ CleanupInterval time.Duration

+ InstanceID string

+ MaxAge time.Duration

+ MaxEvents int

+ type MemoryProviderStats struct

+ ActiveSubscribers atomic.Int32

+ CompletedEvents atomic.Int64

+ EventsConsumed atomic.Int64

+ EventsPublished atomic.Int64

+ Evictions atomic.Int64

+ FailedEvents atomic.Int64

+ PendingEvents atomic.Int64

+ ProcessingEvents atomic.Int64

+ TotalEvents atomic.Int64

+ type Options struct

+ BufferSize int

+ InstanceID string

+ Mode ProcessingMode

+ Provider Provider

+ RetryPolicy *RetryPolicy

+ WorkerCount int

+ type ProcessingMode string

+ const ProcessingModeAsync

+ const ProcessingModeSync

+ type Provider interface

+ Close func() error

+ Delete func(ctx context.Context, id string) error

+ Get func(ctx context.Context, id string) (*Event, error)

+ List func(ctx context.Context, filter *EventFilter) ([]*Event, error)

+ Publish func(ctx context.Context, event *Event) error

+ Stats func(ctx context.Context) (*ProviderStats, error)

+ Store func(ctx context.Context, event *Event) error

+ Stream func(ctx context.Context, pattern string) (<-chan *Event, error)

+ UpdateStatus func(ctx context.Context, id string, status EventStatus, errorMsg string) error

+ func NewProviderFromConfig(cfg config.EventBrokerConfig) (Provider, error)

+ type ProviderStats struct

+ ActiveSubscribers int

+ CompletedEvents int64

+ EventsConsumed int64

+ EventsPublished int64

+ FailedEvents int64

+ PendingEvents int64

+ ProcessingEvents int64

+ ProviderSpecific map[string]interface{}

+ ProviderType string

+ TotalEvents int64

+ type RetryPolicy struct

+ BackoffFactor float64

+ InitialDelay time.Duration

+ MaxDelay time.Duration

+ MaxRetries int

+ func DefaultRetryPolicy() *RetryPolicy

+ type SubscriptionID string

+ func Subscribe(pattern string, handler EventHandler) (SubscriptionID, error)