event

Documentation

Overview

Package event. batch provides batch processing capabilities for events.

Package event. event_bus provides the core pub/sub event bus implementation.

Package event. event_types provides a strongly-typed event bus system for decoupling core logic from side effects.

Package event. filter provides event filtering capabilities for conditional handler execution.

Package event. global provides a singleton event bus instance for application-wide use.

Package event. middleware provides composable middleware for event handlers.

Package event. priority provides priority-based event processing.

Index

• Constants
• func InitGlobalBus(cfg *Config)
• func Publish(event *Event)
• func ShutdownGlobalBus()
• func Subscribe(eventType EventType, handler EventHandler) string
• type BatchConfig
• type BatchHandler
• type BatchProcessor
  • func NewBatchProcessor(eventType EventType, config BatchConfig, handler BatchHandler) *BatchProcessor
  • func (bp *BatchProcessor) Add(event *Event)
  • func (bp *BatchProcessor) AsEventHandler() EventHandler
  • func (bp *BatchProcessor) Flush()
  • func (bp *BatchProcessor) Shutdown()
• type Config
  • func DefaultConfig() *Config
• type Event
  • func NewEvent(eventType EventType, data any) *Event
  • func (e *Event) WithMetadata(key string, value any) *Event
• type EventBus
  • func GetGlobalBus() *EventBus
  • func NewEventBus(cfg *Config) *EventBus
  • func (eb *EventBus) Publish(event *Event)
  • func (eb *EventBus) PublishSync(ctx context.Context, event *Event) []error
  • func (eb *EventBus) Shutdown()
  • func (eb *EventBus) Subscribe(eventType EventType, handler EventHandler) string
  • func (eb *EventBus) Unsubscribe(subscriptionID string) bool
• type EventHandler
  • func NewFilteredHandler(filter Filter, handler EventHandler) EventHandler
• type EventType
• type Filter
  • func And(filters ...Filter) Filter
  • func FilterByMetadata(key string, value interface{}) Filter
  • func FilterByMetadataExists(key string) Filter
  • func Not(filter Filter) Filter
  • func Or(filters ...Filter) Filter
• type FilteredHandler
• type LoginFailedData
• type LoginSuccessData
• type Middleware
  • func Chain(middlewares ...Middleware) Middleware
  • func WithLogging() Middleware
  • func WithRecovery() Middleware
  • func WithRetry(maxRetries int, initialDelay time.Duration) Middleware
  • func WithTimeout(timeout time.Duration) Middleware
• type PriorityEvent
• type PriorityQueue
  • func NewPriorityQueue() *PriorityQueue
  • func (pq *PriorityQueue) Dequeue() (*Event, bool)
  • func (pq *PriorityQueue) Enqueue(event *Event, priority int)
  • func (pq *PriorityQueue) Size() int
• type Subscription
• type TokenRevokedData
• type UserCreatedData
• type UserUpdatedData

Constants

const (
	PriorityLow    = 1
	PriorityNormal = 5
	PriorityHigh   = 10
)

Priority levels for events.

Functions

func InitGlobalBus

func InitGlobalBus(cfg *Config)

InitGlobalBus initializes the global event bus with a custom configuration. IMPORTANT: This function should be called once at the start of your application, before any goroutines that might call GetGlobalBus are started. It has no effect if the global bus has already been initialized (e.g., by a call to GetGlobalBus).

func Publish

func Publish(event *Event)

Publish is a convenience function to publish events to the global bus.

func ShutdownGlobalBus

func ShutdownGlobalBus()

ShutdownGlobalBus shuts down the global event bus.

func Subscribe

func Subscribe(eventType EventType, handler EventHandler) string

Subscribe is a convenience function to subscribe to the global bus.

Types

type BatchConfig

type BatchConfig struct {
	BatchSize   int           // Max events per batch
	FlushPeriod time.Duration // Max time to wait before flushing
}

BatchConfig holds configuration for batch processing.

type BatchHandler

type BatchHandler func(ctx context.Context, events []*Event) error

BatchHandler processes multiple events at once.

type BatchProcessor

type BatchProcessor struct {
	// contains filtered or unexported fields
}

BatchProcessor collects events and processes them in batches.

func NewBatchProcessor

func NewBatchProcessor(eventType EventType, config BatchConfig, handler BatchHandler) *BatchProcessor

NewBatchProcessor creates a new batch processor.

func (*BatchProcessor) Add

func (bp *BatchProcessor) Add(event *Event)

Add adds an event to the batch buffer.

func (*BatchProcessor) AsEventHandler

func (bp *BatchProcessor) AsEventHandler() EventHandler

AsEventHandler returns an EventHandler that adds events to the batch.

func (*BatchProcessor) Flush

func (bp *BatchProcessor) Flush()

Flush processes all buffered events immediately.

func (*BatchProcessor) Shutdown

func (bp *BatchProcessor) Shutdown()

Shutdown stops the batch processor and flushes remaining events.

type Config

type Config struct {
	Workers    int // Number of worker goroutines
	BufferSize int // Size of the event channel buffer
}

Config holds configuration for the event bus.

func DefaultConfig

func DefaultConfig() *Config

DefaultConfig returns sensible default configuration.

type Event

type Event struct {
	Type      EventType      // Strongly-typed event identifier
	Timestamp time.Time      // When the event occurred
	Data      any            // Event-specific payload
	Metadata  map[string]any // Additional context (e.g., request ID, user agent)
}

Event represents a generic event that flows through the system.

func NewEvent

func NewEvent(eventType EventType, data any) *Event

NewEvent creates a new event with the given type and data.

func (*Event) WithMetadata

func (e *Event) WithMetadata(key string, value any) *Event

WithMetadata adds metadata to the event and returns the event for chaining.

type EventBus

type EventBus struct {
	// contains filtered or unexported fields
}

EventBus manages event publishing and subscription.

func GetGlobalBus

func GetGlobalBus() *EventBus

GetGlobalBus returns the singleton event bus instance. On the first call, it initializes the bus with the DefaultConfig configuration if it hasn't been initialized by InitGlobalBus already. Subsequent calls will return the same instance.

func NewEventBus

func NewEventBus(cfg *Config) *EventBus

NewEventBus creates a new event bus with the given configuration.

func (*EventBus) Publish

func (eb *EventBus) Publish(event *Event)

Publish sends an event to all registered handlers asynchronously.

func (*EventBus) PublishSync

func (eb *EventBus) PublishSync(ctx context.Context, event *Event) []error

PublishSync sends an event to all registered handlers synchronously.

func (*EventBus) Shutdown

func (eb *EventBus) Shutdown()

Shutdown gracefully stops the event bus and waits for all workers to finish.

func (*EventBus) Subscribe

func (eb *EventBus) Subscribe(eventType EventType, handler EventHandler) string

Subscribe registers a handler for a specific event type and returns a subscription ID.

func (*EventBus) Unsubscribe

func (eb *EventBus) Unsubscribe(subscriptionID string) bool

Unsubscribe removes a subscription by ID.

type EventHandler

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

EventHandler is a function that processes events.

func NewFilteredHandler

func NewFilteredHandler(filter Filter, handler EventHandler) EventHandler

NewFilteredHandler creates a handler that only executes when the filter returns true.

type EventType

type EventType string

EventType represents a strongly-typed event identifier to prevent typos and ensure consistency.

const (
	// Authentication Events
	AuthEventUserCreated     EventType = "auth.user.created"
	AuthEventLoginSuccess    EventType = "auth.login.success"
	AuthEventLoginFailed     EventType = "auth.login.failed"
	AuthEventLogout          EventType = "auth.logout"
	AuthEventPasswordChanged EventType = "auth.password.changed"
	AuthEventTokenRevoked    EventType = "auth.token.revoked"

	// User Management Events
	UserEventUpdated  EventType = "user.updated"
	UserEventDeleted  EventType = "user.deleted"
	UserEventEnabled  EventType = "user.enabled"
	UserEventDisabled EventType = "user.disabled"

	// Organization Events
	OrgEventCreated EventType = "org.created"
	OrgEventUpdated EventType = "org.updated"
	OrgEventDeleted EventType = "org.deleted"

	// Permission Events
	PermissionEventGranted EventType = "permission.granted"
	PermissionEventRevoked EventType = "permission.revoked"
)

Event type constants for different domains in the system. This is intended for example you can extend it as you want. Simply wrap your string in EventType.

type Filter

type Filter func(*Event) bool

Filter determines if an event should be processed by a handler.

func And

func And(filters ...Filter) Filter

And combines multiple filters with AND logic.

func FilterByMetadata

func FilterByMetadata(key string, value interface{}) Filter

FilterByMetadata creates a filter that checks for specific metadata key-value pairs.

func FilterByMetadataExists

func FilterByMetadataExists(key string) Filter

FilterByMetadataExists creates a filter that checks if metadata key exists.

func Not

func Not(filter Filter) Filter

Not negates a filter.

func Or

func Or(filters ...Filter) Filter

Or combines multiple filters with OR logic.

type FilteredHandler

type FilteredHandler struct {
	// contains filtered or unexported fields
}

FilteredHandler wraps a handler with a filter condition.

type LoginFailedData

type LoginFailedData struct {
	Username  string
	IPAddress string
	Reason    string
}

LoginFailedData contains information about a failed login attempt.

type LoginSuccessData

type LoginSuccessData struct {
	UserID       string
	Username     string
	Organization string
	IPAddress    string
	UserAgent    string
}

LoginSuccessData contains information about a successful login.

type Middleware

type Middleware func(EventHandler) EventHandler

Middleware wraps an EventHandler with additional behavior.

func Chain

func Chain(middlewares ...Middleware) Middleware

Chain chains multiple middlewares together.

func WithLogging

func WithLogging() Middleware

WithLogging logs handler execution.

func WithRecovery

func WithRecovery() Middleware

WithRecovery recovers from panics in handlers.

func WithRetry

func WithRetry(maxRetries int, initialDelay time.Duration) Middleware

WithRetry retries failed handlers with exponential backoff.

func WithTimeout

func WithTimeout(timeout time.Duration) Middleware

WithTimeout adds a timeout to handler execution.

type PriorityEvent

type PriorityEvent struct {
	Event    *Event
	Priority int
	// contains filtered or unexported fields
}

PriorityEvent wraps an event with priority information.

type PriorityQueue

type PriorityQueue struct {
	// contains filtered or unexported fields
}

PriorityQueue manages priority-based event processing.

func NewPriorityQueue

func NewPriorityQueue() *PriorityQueue

NewPriorityQueue creates a new priority queue.

func (*PriorityQueue) Dequeue

func (pq *PriorityQueue) Dequeue() (*Event, bool)

Dequeue removes and returns the highest priority event.

func (*PriorityQueue) Enqueue

func (pq *PriorityQueue) Enqueue(event *Event, priority int)

Enqueue adds an event with the specified priority.

func (*PriorityQueue) Size

func (pq *PriorityQueue) Size() int

Size returns the number of events in the queue.

type Subscription

type Subscription struct {
	ID      string
	Type    EventType
	Handler EventHandler
}

Subscription represents a registered event handler.

type TokenRevokedData

type TokenRevokedData struct {
	TokenID   string
	UserID    string
	RevokedBy string
	Reason    string
}

TokenRevokedData contains information about a revoked token.

type UserCreatedData

type UserCreatedData struct {
	UserID       string
	Username     string
	Email        string
	Organization string
}

UserCreatedData contains information about a newly created user.

type UserUpdatedData

type UserUpdatedData struct {
	UserID        string
	Username      string
	ChangedFields []string
	UpdatedBy     string
}

UserUpdatedData contains information about user profile updates.