Documentation
¶
Index ¶
Constants ¶
const ( // Workers is the configuration key for the number of worker goroutines Workers = "token.finality.notification.workers" // QueueSize is the configuration key for the size of the event buffer QueueSize = "token.finality.notification.queueSize" // DefaultWorkers is the default number of worker goroutines DefaultWorkers = 10 // DefaultQueueSize is the default size of the event buffer DefaultQueueSize = 1000 )
Variables ¶
var ( // ErrQueueClosed is returned when an event is added to a closed queue ErrQueueClosed = errors.New("queue is closed") // ErrQueueFull is returned when a non-blocking enqueue fails because the queue is full ErrQueueFull = errors.New("queue is full") // ErrShutdownTimeout is returned when the shutdown timeout is exceeded ErrShutdownTimeout = errors.New("shutdown timeout exceeded") )
Functions ¶
func NewConfig ¶
func NewConfig(configuration Configuration) *serviceConfig
NewConfig creates a new ConfigGetter that uses the provided Configuration interface to retrieve event queue settings.
Types ¶
type Config ¶
type Config struct {
Workers int // Number of worker goroutines
QueueSize int // Size of the event buffer
MetricsProvider metrics.Provider // Optional metrics provider; uses noop when nil
}
Config holds configuration for the EventQueue
type ConfigGetter ¶
type ConfigGetter interface {
// Workers returns the number of worker goroutines
Workers() int
// QueueSize returns the size of the event buffer
QueueSize() int
}
ConfigGetter models the configuration getter for the event queue
type Configuration ¶
type EventQueue ¶
type EventQueue struct {
// contains filtered or unexported fields
}
EventQueue manages a pool of workers processing events
func NewEventQueue ¶
func NewEventQueue(cfg Config) (*EventQueue, error)
NewEventQueue creates and starts a new event queue with the specified number of workers and buffer size. It validates that both values are greater than 0 before starting the worker pool.
func (*EventQueue) Enqueue ¶
func (eq *EventQueue) Enqueue(event Event) error
Enqueue adds an event to the queue in a non-blocking manner. If the queue is full, it immediately returns ErrQueueFull.
func (*EventQueue) EnqueueBlocking ¶
func (eq *EventQueue) EnqueueBlocking(ctx context.Context, event Event) error
EnqueueBlocking adds an event to the queue, blocking until space is available, the context is canceled, or the queue is closed.
func (*EventQueue) Shutdown ¶
func (eq *EventQueue) Shutdown(timeout time.Duration) error
Shutdown gracefully stops the event queue by setting the closed flag, signaling workers to stop, and waiting for them to finish within the timeout.
func (*EventQueue) Stats ¶
func (eq *EventQueue) Stats() Stats
Stats returns current statistics about the EventQueue, including the number of workers, buffer size, pending events, and closed status.
type Metrics ¶
type Metrics struct {
// PendingEvents is a gauge tracking how many events are currently waiting
// in the queue buffer. Updated on every successful enqueue.
PendingEvents metrics.Gauge
// EnqueueDrops counts the total number of events dropped because the queue
// was full at the time of a non-blocking Enqueue call.
EnqueueDrops metrics.Counter
// ProcessingErrors counts the total number of errors returned by
// event.Process inside worker goroutines.
ProcessingErrors metrics.Counter
// ProcessingDuration is a histogram of successful event processing times in
// worker goroutines, measured in seconds. Only recorded on success; error
// paths are already counted by ProcessingErrors.
ProcessingDuration metrics.Histogram
}
Metrics holds the instrumentation for the EventQueue.
type Stats ¶
type Stats struct {
// Workers is the number of worker goroutines
Workers int
// QueueSize is the size of the event buffer
QueueSize int
// Pending is the number of pending events in the queue
Pending int
// IsClosed is true if the queue is closed
IsClosed bool
}
Stats represents statistics about the event queue
Source Files
Directories