shared

Documentation

Overview

Package shared provides shared types and utilities for queue adapters including configuration, backpressure, circuit breaker, and priority routing.

Index

• type BackpressureManager
  • func NewBackpressureManager(maxQueueSize int, maxMemoryMB int64) *BackpressureManager
  • func (bpm *BackpressureManager) Disable()
  • func (bpm *BackpressureManager) Enable()
  • func (bpm *BackpressureManager) GetCurrentMemoryMB() int64
  • func (bpm *BackpressureManager) IsMemoryExceeded() bool
  • func (bpm *BackpressureManager) ShouldReject(currentQueueSize int) bool
  • func (bpm *BackpressureManager) UpdateMemoryUsage() int64
• type CircuitBreaker
  • func NewCircuitBreaker(threshold float64, minSamples int, recoveryTime time.Duration) *CircuitBreaker
  • func (cb *CircuitBreaker) Call(fn func() error) error
  • func (cb *CircuitBreaker) IsOpen() bool
• type CircuitState
• type PriorityRouter
  • func NewPriorityRouter(config QueueConfig) *PriorityRouter
  • func (pr *PriorityRouter) EstimateComplexity(job *entity.Job) int
  • func (pr *PriorityRouter) Route(job *entity.Job) entity.Priority
  • func (pr *PriorityRouter) ShouldUseFastPath(job *entity.Job) bool
• type QueueConfig
  • func DefaultConfig() QueueConfig

Types

type BackpressureManager

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

BackpressureManager monitors system resources and applies backpressure

func NewBackpressureManager

func NewBackpressureManager(maxQueueSize int, maxMemoryMB int64) *BackpressureManager

NewBackpressureManager creates a new public instance of backpressure manager

func (*BackpressureManager) Disable

func (bpm *BackpressureManager) Disable()

Disable temporarily disables backpressure checks

func (*BackpressureManager) Enable

func (bpm *BackpressureManager) Enable()

Enable allows backpressure checks to be applied

func (*BackpressureManager) GetCurrentMemoryMB

func (bpm *BackpressureManager) GetCurrentMemoryMB() int64

GetCurrentMemoryMB returns current memory usage in MB

func (*BackpressureManager) IsMemoryExceeded

func (bpm *BackpressureManager) IsMemoryExceeded() bool

IsMemoryExceeded checks if memory usage exceeds limit

func (*BackpressureManager) ShouldReject

func (bpm *BackpressureManager) ShouldReject(currentQueueSize int) bool

ShouldReject determines if new jobs should be rejected due to resource constraints

func (*BackpressureManager) UpdateMemoryUsage

func (bpm *BackpressureManager) UpdateMemoryUsage() int64

UpdateMemoryUsage reads current memory stats

type CircuitBreaker

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

CircuitBreaker implements circuit breaker pattern for resilience

func NewCircuitBreaker

func NewCircuitBreaker(threshold float64, minSamples int, recoveryTime time.Duration) *CircuitBreaker

NewCircuitBreaker creates a new public circuit breaker instance

func (*CircuitBreaker) Call

func (cb *CircuitBreaker) Call(fn func() error) error

Call executes function within circuit breaker protection (public wrapper)

func (*CircuitBreaker) IsOpen

func (cb *CircuitBreaker) IsOpen() bool

IsOpen checks if circuit is currently open

type CircuitState

type CircuitState int

CircuitState represents circuit breaker state machine

const (
	StateClosed CircuitState = iota
	StateOpen
	StateHalfOpen
)

type PriorityRouter

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

PriorityRouter routes jobs to appropriate queues based on priority and complexity

func NewPriorityRouter

func NewPriorityRouter(config QueueConfig) *PriorityRouter

NewPriorityRouter creates a new public priority router with given config

func (*PriorityRouter) EstimateComplexity

func (pr *PriorityRouter) EstimateComplexity(job *entity.Job) int

EstimateComplexity provides a rough estimate of job complexity (public wrapper)

func (*PriorityRouter) Route

func (pr *PriorityRouter) Route(job *entity.Job) entity.Priority

Route determines the actual priority for a job (public wrapper)

func (*PriorityRouter) ShouldUseFastPath

func (pr *PriorityRouter) ShouldUseFastPath(job *entity.Job) bool

ShouldUseFastPath determines if job should bypass queueing (public wrapper)

type QueueConfig

type QueueConfig struct {
	// Capacity and workers
	QueueCapacity int
	MinWorkers    int
	MaxWorkers    int

	// Complexity-based routing
	EnableComplexityRouting bool
	SimpleJobThresholdMs    int
	MediumJobThresholdMs    int

	// Fast path
	EnableFastPath      bool
	FastPathThresholdMs int

	// Backpressure management
	EnableBackpressure bool
	MaxMemoryMB        int64

	// Circuit breaker
	EnableCircuitBreaker       bool
	CircuitBreakerThreshold    float64
	CircuitBreakerMinSamples   int
	CircuitBreakerRecoveryTime int

	// Redis-specific (for redis adapter)
	// RedisAddr is used only when RedisClient is nil — creates a standalone connection.
	RedisAddr string
	QueueName string

	// RedisClient is an optional pre-configured redis.UniversalClient.
	// When set, RedisAddr is ignored. Supports standalone, cluster, and sentinel.
	RedisClient goredis.UniversalClient
}

QueueConfig holds configuration shared across all queue implementations

func DefaultConfig

func DefaultConfig() QueueConfig

DefaultConfig returns default queue configuration