mq

package module

Go to main page

Versions in this module

v0

v0.0.47

Oct 7, 2025

v0.0.46

Oct 6, 2025

v0.0.45

Sep 26, 2025

v0.0.43

Sep 20, 2025

v0.0.41

Sep 12, 2025

v0.0.40

Sep 11, 2025

v0.0.38

Sep 11, 2025

v0.0.36

Sep 8, 2025

v0.0.35

Sep 7, 2025

v0.0.33

Sep 5, 2025

v0.0.31

Sep 5, 2025

v0.0.30

Sep 5, 2025

v0.0.26

Sep 3, 2025

v0.0.24

Aug 19, 2025

v0.0.23

Aug 16, 2025

v0.0.22

Aug 12, 2025

v0.0.20

Aug 11, 2025

v0.0.19

Aug 11, 2025

v0.0.18

Aug 10, 2025

v0.0.17

Aug 6, 2025

Changes in this version

+ type AdminBrokerInfo struct

+ Address string

+ Config map[string]interface{}

+ Connections int

+ Status string

+ Uptime int64

+ type AdminConsumerMetrics struct

+ ErrorCount int64

+ ID string

+ LastActivity time.Time

+ MaxConcurrentTasks int

+ MaxRetries int

+ ProcessedTasks int64

+ Queue string

+ Status string

+ TaskTimeout int

+ type AdminHealthCheck struct

+ Duration time.Duration

+ Message string

+ Name string

+ Status string

+ Timestamp time.Time

+ type AdminMessage struct

+ Data interface{}

+ Timestamp time.Time

+ Type string

+ type AdminMetrics struct

+ ActiveConsumers int

+ ActiveQueues int

+ ConsumerMetrics map[string]*AdminConsumerMetrics

+ ErrorCount int64

+ FailedMessages int64

+ PoolMetrics map[string]*AdminPoolMetrics

+ QueueMetrics map[string]*QueueMetrics

+ StartTime time.Time

+ SuccessCount int64

+ SystemMetrics *AdminSystemMetrics

+ ThroughputHistory []float64

+ TotalMessages int64

+ func NewAdminMetrics() *AdminMetrics

+ type AdminPoolMetrics struct

+ ActiveTasks int

+ ID string

+ LastActivity time.Time

+ MaxMemoryLoad int64

+ QueueSize int

+ Status string

+ Workers int

+ type AdminQueueInfo struct

+ Consumers int

+ Depth int

+ Name string

+ Rate int

+ type AdminServer struct

+ func NewAdminServer(broker *Broker, addr string, log logger.Logger) *AdminServer

+ func (a *AdminServer) BroadcastTaskUpdate(update *TaskUpdate)

+ func (a *AdminServer) Start() error

+ func (a *AdminServer) Stop() error

+ type AdminSystemMetrics struct

+ CPUPercent float64

+ GoroutineCount int

+ MemoryPercent float64

+ Timestamp time.Time

type Broker

+ func (b *Broker) GetConsumers() []*AdminConsumerMetrics

+ func (b *Broker) GetPools() []*AdminPoolMetrics

+ func (b *Broker) UpdateConsumerMetrics(consumerID string, processedTasks, errorCount int64)

+ func (b *Broker) UpdateConsumerPool(consumerID string, pool *Pool)

+ type ConsumerMetrics struct

+ ErrorCount int64

+ LastActivity time.Time

+ ProcessedTasks int64

type DeadLetterQueue

+ func (dlq *DeadLetterQueue) Clear() int

+ func (dlq *DeadLetterQueue) GetStats() map[string]interface{}

+ func (dlq *DeadLetterQueue) GetTasksByErrorType(errorPattern string) []*QueueTask

+ func (dlq *DeadLetterQueue) RemoveOlderThan(duration time.Duration) int

+ func (dlq *DeadLetterQueue) Size() int

type Pool

+ func (wp *Pool) FlushQueues() error

+ func (wp *Pool) GetCurrentConfig() DynamicConfig

+ func (wp *Pool) GetHealthStatus() PoolHealthStatus

+ func (wp *Pool) GetOverflowDepth() int

+ func (wp *Pool) GetQueueDepth() int

+ func (wp *Pool) PauseProcessing()

+ func (wp *Pool) RecoverFromFailure() error

+ func (wp *Pool) ResumeProcessing()

+ type PoolHealthStatus struct

+ CircuitBreakerOpen bool

+ DLQDepth int

+ ErrorRate float64

+ IsHealthy bool

+ Issues []string

+ LastTaskProcessedAt *time.Time

+ MemoryUsage string

+ MemoryUsagePercent float64

+ OverflowDepth int

+ QueueDepth int

+ ThroughputPerSecond float64

+ Uptime time.Duration

+ WorkerCount int32

type QueueMetrics

+ ActiveTasks int

+ MaxMemoryLoad int64

+ QueueDepth int

+ WorkerCount int

+ type TaskUpdate struct

+ Consumer string

+ CreatedAt time.Time

+ Error string

+ Queue string

+ Status string

+ TaskID string

+ UpdatedAt time.Time

v0.0.16

Jul 29, 2025

Changes in this version

+ type ActiveAlert struct

+ Annotations map[string]string

+ EndsAt *time.Time

+ Labels map[string]string

+ Rule AlertRule

+ StartsAt time.Time

+ Status AlertStatus

+ Value float64

+ type AlertManager struct

+ func NewAlertManager(logger logger.Logger) *AlertManager

+ func (am *AlertManager) AddNotifier(notifier AlertNotifier)

+ func (am *AlertManager) AddRule(rule AlertRule)

+ func (am *AlertManager) EvaluateRules(registry *DetailedMetricsRegistry)

+ type AlertNotifier interface

+ Name func() string

+ Notify func(ctx context.Context, alert ActiveAlert) error

+ type AlertRule struct

+ Annotations map[string]string

+ Condition string

+ Duration time.Duration

+ Enabled bool

+ Labels map[string]string

+ Metric string

+ Name string

+ Threshold float64

+ type AlertStatus string

+ const AlertStatusFiring

+ const AlertStatusResolved

+ const AlertStatusSilenced

type Broker

+ func (b *Broker) Authenticate(ctx context.Context, credentials map[string]string) error

+ func (b *Broker) Authorize(ctx context.Context, role string, action string) error

+ func (b *Broker) NewQueueWithConfig(name string, opts ...QueueOption) *Queue

+ func (b *Broker) NewQueueWithOrdering(name string) *Queue

+ func (b *Broker) ReprocessDLQ(queueName string) error

+ func (b *Broker) StartEnhanced(ctx context.Context) error

+ func (b *Broker) StopEnhanced() error

+ type BrokerConfig struct

+ Address string

+ ConnectionTimeout time.Duration

+ DeadLetterMaxRetries int

+ EnableDeadLetter bool

+ EnableMetrics bool

+ GracefulShutdown time.Duration

+ Headers map[string]string

+ IdleTimeout time.Duration

+ KeepAlive bool

+ KeepAlivePeriod time.Duration

+ MaxConnections int

+ MaxQueueDepth int

+ MessageTTL time.Duration

+ MetricsInterval time.Duration

+ Port int

+ ReadTimeout time.Duration

+ WriteTimeout time.Duration

+ func (b *BrokerConfig) UnmarshalJSON(data []byte) error

+ type BrokerConnection struct

+ type CircuitBreaker struct

+ Enabled bool

+ FailureThreshold int

+ ResetTimeout time.Duration

+ type CircuitState int

+ const CircuitClosed

+ const CircuitHalfOpen

+ const CircuitOpen

+ type ClusteringConfig struct

+ ClusterNodes []string

+ ConsistencyLevel string

+ DiscoveryEndpoint string

+ DiscoveryMethod string

+ ElectionTimeout time.Duration

+ EnableClustering bool

+ EnableFailover bool

+ EnableLoadBalancing bool

+ EnableReplication bool

+ FailoverTimeout time.Duration

+ HeartbeatInterval time.Duration

+ LoadBalancingStrategy string

+ NodeID string

+ ReplicationFactor int

+ func (c *ClusteringConfig) UnmarshalJSON(data []byte) error

+ type ConfigManager struct

+ func NewConfigManager(configFile string, logger logger.Logger) *ConfigManager

+ func (cm *ConfigManager) AddWatcher(watcher ConfigWatcher)

+ func (cm *ConfigManager) GetConfig() *ProductionConfig

+ func (cm *ConfigManager) LoadConfig() error

+ func (cm *ConfigManager) RemoveWatcher(watcher ConfigWatcher)

+ func (cm *ConfigManager) SaveConfig() error

+ func (cm *ConfigManager) StartWatching(ctx context.Context, interval time.Duration)

+ func (cm *ConfigManager) UpdateConfig(newConfig *ProductionConfig) error

+ type ConfigWatcher interface

+ OnConfigChange func(oldConfig, newConfig *ProductionConfig) error

+ type ConnectionPool struct

+ func NewConnectionPool(maxConns int) *ConnectionPool

+ func (cp *ConnectionPool) AddConnection(id string, conn net.Conn, connType string) error

+ func (cp *ConnectionPool) CleanupIdleConnections(idleTimeout time.Duration)

+ func (cp *ConnectionPool) GetActiveConnections() int64

+ func (cp *ConnectionPool) RemoveConnection(id string)

+ type ConsumerConfig struct

+ CircuitBreakerThreshold int

+ CircuitBreakerTimeout time.Duration

+ DeduplicationWindow time.Duration

+ EnableCircuitBreaker bool

+ EnableDeduplication bool

+ EnableHTTPAPI bool

+ EnablePriorityQueue bool

+ EnableReconnect bool

+ HTTPAPIPort int

+ HealthCheckInterval time.Duration

+ InitialDelay time.Duration

+ JitterPercent float64

+ MaxBackoff time.Duration

+ MaxConcurrentTasks int

+ MaxRetries int

+ ReconnectInterval time.Duration

+ TaskTimeout time.Duration

+ func (c *ConsumerConfig) UnmarshalJSON(data []byte) error

type CronSchedule

+ Seconds string

+ type DetailedMetricsRegistry struct

+ func NewDetailedMetricsRegistry() *DetailedMetricsRegistry

+ func (mr *DetailedMetricsRegistry) GetAllMetrics() map[string]*TimeSeries

+ func (mr *DetailedMetricsRegistry) GetMetric(name string) (*TimeSeries, bool)

+ func (mr *DetailedMetricsRegistry) RecordValue(name string, value float64)

+ func (mr *DetailedMetricsRegistry) RegisterMetric(name string, metricType MetricType, description string, ...)

+ type DiskSpaceHealthCheck struct

+ func (dshc *DiskSpaceHealthCheck) Check(ctx context.Context) *HealthCheckResult

+ func (dshc *DiskSpaceHealthCheck) Name() string

+ func (dshc *DiskSpaceHealthCheck) Timeout() time.Duration

+ type DistributedLocker interface

+ Acquire func(key string, ttl time.Duration) bool

+ Release func(key string)

+ type EnhancedCircuitBreaker struct

+ func NewEnhancedCircuitBreaker(threshold int64, timeout time.Duration) *EnhancedCircuitBreaker

+ func (cb *EnhancedCircuitBreaker) Call(fn func() error) error

+ type GoRoutineHealthCheck struct

+ func (ghc *GoRoutineHealthCheck) Check(ctx context.Context) *HealthCheckResult

+ func (ghc *GoRoutineHealthCheck) Name() string

+ func (ghc *GoRoutineHealthCheck) Timeout() time.Duration

+ type HealthCheck interface

+ Check func(ctx context.Context) *HealthCheckResult

+ Name func() string

+ Timeout func() time.Duration

+ type HealthCheckResult struct

+ Duration time.Duration

+ Message string

+ Metadata map[string]interface{}

+ Name string

+ Status HealthStatus

+ Timestamp time.Time

+ type HealthChecker struct

+ func NewHealthChecker() *HealthChecker

+ func (hc *HealthChecker) Start()

+ func (hc *HealthChecker) Stop()

+ type HealthStatus string

+ const HealthStatusHealthy

+ const HealthStatusUnhealthy

+ const HealthStatusUnknown

+ const HealthStatusWarning

+ type HealthThresholds struct

+ MaxCPUUsage float64

+ MaxConnections int

+ MaxMemoryUsage int64

+ MaxQueueDepth int

+ MaxResponseTime time.Duration

+ MinFreeMemory int64

+ type InMemoryMessageStore struct

+ func NewInMemoryMessageStore() *InMemoryMessageStore

+ func (ims *InMemoryMessageStore) Cleanup(olderThan time.Time) error

+ func (ims *InMemoryMessageStore) Count(queue string) (int64, error)

+ func (ims *InMemoryMessageStore) Delete(id string) error

+ func (ims *InMemoryMessageStore) List(queue string, limit int, offset int) ([]*StoredMessage, error)

+ func (ims *InMemoryMessageStore) Retrieve(id string) (*StoredMessage, error)

+ func (ims *InMemoryMessageStore) Store(msg *StoredMessage) error

+ type LockEntry struct

+ type LogNotifier struct

+ func NewLogNotifier(logger logger.Logger) *LogNotifier

+ func (ln *LogNotifier) Name() string

+ func (ln *LogNotifier) Notify(ctx context.Context, alert ActiveAlert) error

+ type MemoryHealthCheck struct

+ func (mhc *MemoryHealthCheck) Check(ctx context.Context) *HealthCheckResult

+ func (mhc *MemoryHealthCheck) Name() string

+ func (mhc *MemoryHealthCheck) Timeout() time.Duration

+ type MessageStore interface

+ Cleanup func(olderThan time.Time) error

+ Count func(queue string) (int64, error)

+ Delete func(id string) error

+ List func(queue string, limit int, offset int) ([]*StoredMessage, error)

+ Retrieve func(id string) (*StoredMessage, error)

+ Store func(msg *StoredMessage) error

+ type Metric struct

+ Name string

+ Tags map[string]string

+ Timestamp time.Time

+ Value float64

+ type MetricType string

+ const MetricTypeCounter

+ const MetricTypeGauge

+ const MetricTypeHistogram

+ const MetricTypeSummary

+ type MetricsCollector struct

+ func NewMetricsCollector() *MetricsCollector

+ func (mc *MetricsCollector) RecordMetric(name string, value float64, tags map[string]string)

+ type MetricsServer struct

+ func NewMetricsServer(broker *Broker, config *MonitoringConfig, logger logger.Logger) *MetricsServer

+ func (ms *MetricsServer) AddAlertNotifier(notifier AlertNotifier)

+ func (ms *MetricsServer) AddAlertRule(rule AlertRule)

+ func (ms *MetricsServer) Start(ctx context.Context) error

+ func (ms *MetricsServer) Stop() error

+ type MonitoringConfig struct

+ EnableHealthCheck bool

+ EnableLogging bool

+ EnableMetrics bool

+ EnableProfiling bool

+ EnableTracing bool

+ HealthCheckInterval time.Duration

+ HealthCheckPath string

+ HealthCheckPort int

+ LogFilePath string

+ LogFormat string

+ LogLevel string

+ LogMaxAge int

+ LogMaxBackups int

+ LogMaxSize int

+ LogOutput string

+ MetricsPath string

+ MetricsPort int

+ ProfilingPort int

+ TracingEndpoint string

+ TracingSampleRate float64

+ func (m *MonitoringConfig) UnmarshalJSON(data []byte) error

type Option

+ func WithConsumerTimeout(timeout time.Duration) Option

type Options

+ func (o *Options) ConsumerTimeout() time.Duration

+ type PersistenceConfig struct

+ BackupEnabled bool

+ BackupInterval time.Duration

+ BackupPath string

+ CleanupInterval time.Duration

+ CompressionEnabled bool

+ ConnectionString string

+ ConnectionTimeout time.Duration

+ EnablePersistence bool

+ EncryptionEnabled bool

+ MaxConnections int

+ ReplicationEnabled bool

+ ReplicationNodes []string

+ RetentionPeriod time.Duration

+ StorageType string

+ func (p *PersistenceConfig) UnmarshalJSON(data []byte) error

type Pool

+ func (wp *Pool) AddScheduledMetrics(total int)

+ func (wp *Pool) Init()

+ type PoolConfig struct

+ EnableDynamicScaling bool

+ EnablePriorityScheduling bool

+ EnableWorkStealing bool

+ GracefulShutdownTimeout time.Duration

+ IdleWorkerTimeout time.Duration

+ MaxMemoryLoad int64

+ MaxQueueWaitTime time.Duration

+ MaxWorkers int

+ MinWorkers int

+ QueueSize int

+ ScalingFactor float64

+ ScalingInterval time.Duration

+ TaskTimeout time.Duration

+ func (p *PoolConfig) UnmarshalJSON(data []byte) error

+ type ProductionConfig struct

+ Broker BrokerConfig

+ Clustering ClusteringConfig

+ Consumer ConsumerConfig

+ LastUpdated time.Time

+ Monitoring MonitoringConfig

+ Persistence PersistenceConfig

+ Pool PoolConfig

+ Publisher PublisherConfig

+ RateLimit RateLimitConfig

+ Security SecurityConfig

+ func DefaultProductionConfig() *ProductionConfig

+ func (c *ProductionConfig) UnmarshalJSON(data []byte) error

+ type PublisherConfig struct

+ AsyncBufferSize int

+ BatchSize int

+ BatchTimeout time.Duration

+ CompressionLevel int

+ ConnectionPoolSize int

+ EnableAsync bool

+ EnableBatching bool

+ EnableCompression bool

+ EnableOrderedDelivery bool

+ InitialDelay time.Duration

+ JitterPercent float64

+ MaxBackoff time.Duration

+ MaxRetries int

+ PublishTimeout time.Duration

+ func (p *PublisherConfig) UnmarshalJSON(data []byte) error

+ type QueueConfig struct

+ BatchSize int

+ CompactionMode bool

+ DeadLetter bool

+ MaxDepth int

+ MaxRetries int

+ MessageTTL time.Duration

+ OrderedMode bool

+ Persistent bool

+ Priority int

+ ThrottleBurst int

+ ThrottleRate int

+ Throttling bool

+ type QueueMetrics struct

+ AverageLatency time.Duration

+ CurrentDepth int64

+ LastActivity time.Time

+ MessagesFailed int64

+ MessagesProcessed int64

+ MessagesReceived int64

+ type QueueOption func(*QueueConfig)

+ func WithDeadLetter() QueueOption

+ func WithPersistent() QueueOption

+ func WithQueueMaxDepth(maxDepth int) QueueOption

+ func WithQueueMaxRetries(maxRetries int) QueueOption

+ func WithQueueOption(config QueueConfig) QueueOption

+ func WithQueueTTL(ttl time.Duration) QueueOption

+ type RateLimitConfig struct

+ BrokerBurst int

+ BrokerRate int

+ ConsumerBurst int

+ ConsumerRate int

+ EnableBrokerRateLimit bool

+ EnableConsumerRateLimit bool

+ EnablePerQueueRateLimit bool

+ EnablePublisherRateLimit bool

+ PerQueueBurst int

+ PerQueueRate int

+ PublisherBurst int

+ PublisherRate int

type ScheduleOptions

+ ScheduleSpec string

type Scheduler

+ func (s *Scheduler) Close() error

+ func (s *Scheduler) ListScheduledTasks() []TaskInfo

+ func (s *Scheduler) UpdateTask(id string, newSched *Schedule) error

type SchedulerConfig

+ MaxRetries int

+ type SchedulerOpt func(*Scheduler)

+ func WithStorage(sm storage.IMap[string, *ScheduledTask]) SchedulerOpt

type SchedulerOption

+ func WithScheduleSpec(spec string) SchedulerOption

+ type SecurityConfig struct

+ AuditLogPath string

+ AuthenticationMethod string

+ EnableAuditLog bool

+ EnableAuthentication bool

+ EnableAuthorization bool

+ EnableEncryption bool

+ EnableTLS bool

+ EncryptionKey string

+ LockoutDuration time.Duration

+ MaxLoginAttempts int

+ SessionTimeout time.Duration

+ TLSCAPath string

+ TLSCertPath string

+ TLSInsecureSkipVerify bool

+ TLSKeyPath string

+ func (s *SecurityConfig) UnmarshalJSON(data []byte) error

+ type StoredMessage struct

+ Attempts int

+ CreatedAt time.Time

+ ExpiresAt *time.Time

+ Headers map[string]string

+ ID string

+ Metadata map[string]interface{}

+ Payload []byte

+ Priority int

+ Queue string

+ type SystemHealthChecker struct

+ func NewSystemHealthChecker(logger logger.Logger) *SystemHealthChecker

+ func (shc *SystemHealthChecker) GetOverallHealth() HealthStatus

+ func (shc *SystemHealthChecker) RegisterCheck(check HealthCheck)

+ func (shc *SystemHealthChecker) RunChecks(ctx context.Context) map[string]*HealthCheckResult

type Task

+ DedupKey string

+ ErrorMsg string

+ Headers map[string]string

+ MaxRetries int

+ Priority int

+ Retries int

+ SpanID string

+ Tags map[string]string

+ TraceID string

+ func (t *Task) AddHeader(key, value string)

+ func (t *Task) AddTag(key, value string)

+ func (t *Task) CanRetry() bool

+ func (t *Task) GetError() error

+ func (t *Task) IncrementRetry()

+ func (t *Task) IsExpired() bool

+ func (t *Task) SetError(err error)

+ type TaskInfo struct

+ NextRunTime time.Time

+ TaskID string

type TaskOption

+ func WithDedupKey(key string) TaskOption

+ func WithExpiry(expiry time.Time) TaskOption

+ func WithPriority(priority int) TaskOption

+ func WithTTL(ttl time.Duration) TaskOption

+ func WithTags(tags map[string]string) TaskOption

+ func WithTaskHeaders(headers map[string]string) TaskOption

+ func WithTaskMaxRetries(maxRetries int) TaskOption

+ func WithTraceID(traceID string) TaskOption

+ type TimeSeries struct

+ Description string

+ Labels map[string]string

+ MaxPoints int

+ Name string

+ Type MetricType

+ Values []TimeSeriesPoint

+ type TimeSeriesPoint struct

+ Timestamp time.Time

+ Value float64

v0.0.15

Apr 7, 2025

v0.0.14

Apr 7, 2025

v0.0.13

Apr 7, 2025

v0.0.12

Mar 30, 2025

Changes in this version

+ var BrokerAddr string

type Consumer

+ func (c *Consumer) StartHTTPAPI() (int, error)

+ type FormattedMetrics struct

+ AverageExecution string

+ CompletedTasks int64

+ CumulativeExecution string

+ CumulativeMemoryUsed string

+ CurrentMemoryUsed string

+ ErrorCount int64

+ TotalScheduled int64

+ TotalTasks int64

type Metrics

+ CumulativeMemoryUsed int64

type Option

+ func WithHTTPApi(flag bool) Option

type Options

+ func (o *Options) HTTPApi() bool

type Pool

+ func (wp *Pool) FormattedMetrics() FormattedMetrics

type RateLimiter

+ func (rl *RateLimiter) Stop()

+ func (rl *RateLimiter) Update(newRate, newBurst int)

v0.0.11

Mar 30, 2025

Changes in this version

+ var Config = &DynamicConfig

+ var Logger = log.DefaultLogger

type Broker

+ func (b *Broker) AdjustConsumerWorkers(noOfWorkers int, consumerID ...string)

+ func (b *Broker) OnConsumerUpdated(ctx context.Context, msg *codec.Message)

+ func (b *Broker) UpdateConsumer(ctx context.Context, consumerID string, config DynamicConfig, queues ...string) error

+ type CircuitBreakerConfig struct

+ Enabled bool

+ FailureThreshold int

+ ResetTimeout time.Duration

type Consumer

+ func (c *Consumer) Update(ctx context.Context, payload []byte) error

+ type DeadLetterQueue struct

+ func NewDeadLetterQueue() *DeadLetterQueue

+ func (dlq *DeadLetterQueue) Add(task *QueueTask)

+ func (dlq *DeadLetterQueue) Tasks() []*QueueTask

+ type DefaultPlugin struct

+ func (dp *DefaultPlugin) AfterTask(task *QueueTask, result Result)

+ func (dp *DefaultPlugin) BeforeTask(task *QueueTask)

+ func (dp *DefaultPlugin) Initialize(config interface{}) error

+ type DynamicConfig struct

+ BackoffDuration time.Duration

+ BatchSize int

+ IdleTimeout time.Duration

+ MaxMemoryLoad int64

+ MaxRetries int

+ NumberOfWorkers int

+ ReloadInterval time.Duration

+ Timeout time.Duration

+ WarningThreshold WarningThresholds

+ type InMemoryMetricsRegistry struct

+ func NewInMemoryMetricsRegistry() *InMemoryMetricsRegistry

+ func (m *InMemoryMetricsRegistry) Get(metricName string) interface{}

+ func (m *InMemoryMetricsRegistry) Increment(metricName string)

+ func (m *InMemoryMetricsRegistry) Register(metricName string, value interface{})

+ type MetricsRegistry interface

+ Get func(metricName string) interface{}

+ Increment func(metricName string)

+ Register func(metricName string, value interface{})

type Option

+ func DisableBrokerRateLimit() Option

+ func DisableConsumerRateLimit() Option

+ func WithBrokerRateLimiter(rate int, burst int) Option

+ func WithConsumerRateLimiter(rate int, burst int) Option

type Options

+ BrokerRateLimiter *RateLimiter

+ ConsumerRateLimiter *RateLimiter

+ type Plugin interface

+ AfterTask func(task *QueueTask, result Result)

+ BeforeTask func(task *QueueTask)

+ Initialize func(config interface{}) error

type Pool

+ func (wp *Pool) DLQ() *DeadLetterQueue

+ func (wp *Pool) UpdateConfig(newConfig *DynamicConfig) error

type PoolOption

+ func WithCircuitBreaker(config CircuitBreakerConfig) PoolOption

+ func WithDiagnostics(enabled bool) PoolOption

+ func WithGracefulShutdown(timeout time.Duration) PoolOption

+ func WithHealthServicePort(port int) PoolOption

+ func WithMetricsRegistry(registry MetricsRegistry) PoolOption

+ func WithPlugin(plugin Plugin) PoolOption

+ func WithWarningThresholds(thresholds ThresholdConfig) PoolOption

+ type RateLimiter struct

+ C chan struct{}

+ func NewRateLimiter(rate int, burst int) *RateLimiter

+ func (rl *RateLimiter) Wait()

type Status

+ const Cancelled

+ type ThresholdConfig struct

+ HighMemory int64

+ LongExecution time.Duration

+ type WarningThresholds struct

+ HighMemory int64

+ LongExecution time.Duration

v0.0.10

Mar 16, 2025

Changes in this version

type Option

+ func WithLogger(log logger.Logger) Option

type Options

+ func (o *Options) BrokerAddr() string

+ func (o *Options) Logger() logger.Logger

type Result

+ Last bool

type Task

+ func (t *Task) GetFlow() any

+ type TaskOption func(*Task)

+ func WithDAG(dag any) TaskOption

v0.0.9

Jan 19, 2025

v0.0.8

Dec 26, 2024

v0.0.7

Nov 30, 2024

v0.0.6

Nov 26, 2024

v0.0.5

Nov 24, 2024

v0.0.4

Nov 24, 2024

v0.0.3

Nov 23, 2024

v0.0.2

Nov 23, 2024

Changes in this version

type Broker

+ func (b *Broker) Close() error

+ func (b *Broker) SetURL(url string)

+ type Status string

+ const Completed

+ const Failed

+ const Pending

+ const Processing

v0.0.1

Nov 2, 2024

Changes in this version

+ func GetAwaitResponse(ctx context.Context) (string, bool)

+ func GetConnection(addr string, config TLSConfig) (net.Conn, error)

+ func GetConsumerID(ctx context.Context) (string, bool)

+ func GetContentType(ctx context.Context) (string, bool)

+ func GetHeader(ctx context.Context, key string) (string, bool)

+ func GetHeaders(ctx context.Context) (storage.IMap[string, string], bool)

+ func GetPublisherID(ctx context.Context) (string, bool)

+ func GetQueue(ctx context.Context) (string, bool)

+ func GetTriggerNode(ctx context.Context) (string, bool)

+ func HeadersWithConsumerID(ctx context.Context, id string) map[string]string

+ func HeadersWithConsumerIDAndQueue(ctx context.Context, id, queue string) map[string]string

+ func IsClosed(conn net.Conn) bool

+ func NewID() string

+ func RecoverPanic(labelGenerator func() string)

+ func RecoverTitle() string

+ func SetHeaders(ctx context.Context, headers map[string]string) context.Context

+ func WithHeaders(ctx context.Context, headers map[string]string) map[string]string

+ func WrapError(err error, msg, op string) error

+ type Broker struct

+ func NewBroker(opts ...Option) *Broker

+ func (b *Broker) AddConsumer(ctx context.Context, queueName string, conn net.Conn) string

+ func (b *Broker) HandleCallback(ctx context.Context, msg *codec.Message)

+ func (b *Broker) MessageAck(ctx context.Context, msg *codec.Message)

+ func (b *Broker) MessageDeny(ctx context.Context, msg *codec.Message)

+ func (b *Broker) MessageResponseHandler(ctx context.Context, msg *codec.Message)

+ func (b *Broker) NewQueue(name string) *Queue

+ func (b *Broker) NotifyHandler() func(context.Context, Result) error

+ func (b *Broker) OnClose(ctx context.Context, conn net.Conn) error

+ func (b *Broker) OnConsumerPause(ctx context.Context, _ *codec.Message)

+ func (b *Broker) OnConsumerResume(ctx context.Context, _ *codec.Message)

+ func (b *Broker) OnConsumerStop(ctx context.Context, _ *codec.Message)

+ func (b *Broker) OnError(_ context.Context, conn net.Conn, err error)

+ func (b *Broker) OnMessage(ctx context.Context, msg *codec.Message, conn net.Conn)

+ func (b *Broker) Options() *Options

+ func (b *Broker) PauseConsumer(ctx context.Context, consumerID string, queues ...string)

+ func (b *Broker) Publish(ctx context.Context, task *Task, queue string) error

+ func (b *Broker) PublishHandler(ctx context.Context, conn net.Conn, msg *codec.Message)

+ func (b *Broker) RemoveConsumer(consumerID string, queues ...string)

+ func (b *Broker) ResumeConsumer(ctx context.Context, consumerID string, queues ...string)

+ func (b *Broker) SetNotifyHandler(callback Callback)

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

+ func (b *Broker) StopConsumer(ctx context.Context, consumerID string, queues ...string)

+ func (b *Broker) SubscribeHandler(ctx context.Context, conn net.Conn, msg *codec.Message)

+ func (b *Broker) SyncMode() bool

+ func (b *Broker) TLSConfig() TLSConfig

+ func (b *Broker) URL() string

+ type Callback func(ctx context.Context, result Result) error

+ type CompletionCallback func()

+ type Consumer struct

+ func NewConsumer(id string, queue string, handler Handler, opts ...Option) *Consumer

+ func (c *Consumer) Close() error

+ func (c *Consumer) Conn() net.Conn

+ func (c *Consumer) Consume(ctx context.Context) error

+ func (c *Consumer) ConsumeMessage(ctx context.Context, msg *codec.Message, conn net.Conn)

+ func (c *Consumer) GetKey() string

+ func (c *Consumer) GetType() string

+ func (c *Consumer) Metrics() Metrics

+ func (c *Consumer) OnClose(_ context.Context, _ net.Conn) error

+ func (c *Consumer) OnError(_ context.Context, conn net.Conn, err error)

+ func (c *Consumer) OnMessage(ctx context.Context, msg *codec.Message, conn net.Conn) error

+ func (c *Consumer) OnResponse(ctx context.Context, result Result) error

+ func (c *Consumer) Pause(ctx context.Context) error

+ func (c *Consumer) ProcessTask(ctx context.Context, msg *Task) Result

+ func (c *Consumer) Resume(ctx context.Context) error

+ func (c *Consumer) SetKey(key string)

+ func (c *Consumer) Stop(ctx context.Context) error

+ type CronSchedule struct

+ DayOfMonth string

+ DayOfWeek string

+ Hour string

+ Minute string

+ Month string

+ func (c CronSchedule) String() string

+ type ExecutionHistory struct

+ Result Result

+ Timestamp time.Time

+ type Handler func(context.Context, *Task) Result

+ type MemoryTaskStorage struct

+ func NewMemoryTaskStorage(expiryTime time.Duration) *MemoryTaskStorage

+ func (m *MemoryTaskStorage) CleanupExpiredTasks() error

+ func (m *MemoryTaskStorage) DeleteTask(taskID string) error

+ func (m *MemoryTaskStorage) FetchNextTask() (*QueueTask, error)

+ func (m *MemoryTaskStorage) GetAllTasks() ([]*QueueTask, error)

+ func (m *MemoryTaskStorage) GetTask(taskID string) (*QueueTask, error)

+ func (m *MemoryTaskStorage) SaveTask(task *QueueTask) error

+ type Metrics struct

+ CompletedTasks int64

+ ErrorCount int64

+ ExecutionTime int64

+ TotalMemoryUsed int64

+ TotalScheduled int64

+ TotalTasks int64

+ type Option func(*Options)

+ func WithBrokerURL(url string) Option

+ func WithCAPath(caPath string) Option

+ func WithCallback(val ...func(context.Context, Result) Result) Option

+ func WithCleanTaskOnComplete() Option

+ func WithConsumerOnClose(handler func(ctx context.Context, topic, consumerName string)) Option

+ func WithConsumerOnSubscribe(handler func(ctx context.Context, topic, consumerName string)) Option

+ func WithInitialDelay(val time.Duration) Option

+ func WithJitterPercent(val float64) Option

+ func WithMaxBackoff(val time.Duration) Option

+ func WithMaxRetries(val int) Option

+ func WithNotifyResponse(callback Callback) Option

+ func WithRespondPendingResult(mode bool) Option

+ func WithSyncMode(mode bool) Option

+ func WithTLS(enableTLS bool, certPath, keyPath string) Option

+ func WithWorkerPool(queueSize, numOfWorkers int, maxMemoryLoad int64) Option

+ type Options struct

+ func SetupOptions(opts ...Option) *Options

+ func (o *Options) CleanTaskOnComplete() bool

+ func (o *Options) MaxMemoryLoad() int64

+ func (o *Options) NumOfWorkers() int

+ func (o *Options) QueueSize() int

+ func (o *Options) SetSyncMode(sync bool)

+ func (o *Options) Storage() TaskStorage

+ type Pool struct

+ func NewPool(numOfWorkers int, opts ...PoolOption) *Pool

+ func (wp *Pool) AdjustWorkerCount(newWorkerCount int)

+ func (wp *Pool) Dispatch(event func())

+ func (wp *Pool) EnqueueTask(ctx context.Context, payload *Task, priority int) error

+ func (wp *Pool) Metrics() Metrics

+ func (wp *Pool) Pause()

+ func (wp *Pool) Resume()

+ func (wp *Pool) Scheduler() *Scheduler

+ func (wp *Pool) SetBatchSize(size int)

+ func (wp *Pool) Start(numWorkers int)

+ func (wp *Pool) Stop()

+ type PoolOption func(*Pool)

+ func WithBatchSize(batchSize int) PoolOption

+ func WithCompletionCallback(callback func()) PoolOption

+ func WithHandler(handler Handler) PoolOption

+ func WithMaxMemoryLoad(maxMemoryLoad int64) PoolOption

+ func WithPoolCallback(callback Callback) PoolOption

+ func WithTaskQueueSize(size int) PoolOption

+ func WithTaskStorage(storage TaskStorage) PoolOption

+ func WithTaskTimeout(t time.Duration) PoolOption

+ type PriorityQueue []*QueueTask

+ func (pq *PriorityQueue) Pop() interface{}

+ func (pq *PriorityQueue) Push(x interface{})

+ func (pq PriorityQueue) Len() int

+ func (pq PriorityQueue) Less(i, j int) bool

+ func (pq PriorityQueue) Swap(i, j int)

+ type Processor interface

+ Close func() error

+ Consume func(ctx context.Context) error

+ GetKey func() string

+ GetType func() string

+ Pause func(ctx context.Context) error

+ ProcessTask func(ctx context.Context, msg *Task) Result

+ Resume func(ctx context.Context) error

+ SetKey func(key string)

+ Stop func(ctx context.Context) error

+ type Publisher struct

+ func NewPublisher(id string, opts ...Option) *Publisher

+ func (p *Publisher) Publish(ctx context.Context, task Task, queue string) error

+ func (p *Publisher) Request(ctx context.Context, queue string, task Task) Result

+ type Queue struct

+ type QueueTask struct

+ type QueuedTask struct

+ Message *codec.Message

+ RetryCount int

+ type Result struct

+ ConditionStatus string

+ CreatedAt time.Time

+ Ctx context.Context

+ Error error

+ Latency string

+ Payload json.RawMessage

+ ProcessedAt time.Time

+ Status string

+ TaskID string

+ Topic string

+ func HandleError(ctx context.Context, err error, status ...string) Result

+ func (r *Result) UnmarshalJSON(data []byte) error

+ func (r Result) MarshalJSON() ([]byte, error)

+ func (r Result) Unmarshal(data any) error

+ func (r Result) WithData(status string, data []byte) Result

+ type Schedule struct

+ CronSpec string

+ DayOfMonth []int

+ DayOfWeek []time.Weekday

+ Interval time.Duration

+ Recurring bool

+ TimeOfDay time.Time

+ func (s *Schedule) ToHumanReadable() string

+ type ScheduleOptions struct

+ Callback Callback

+ Handler Handler

+ Interval time.Duration

+ Overlap bool

+ Recurring bool

+ type ScheduledTask struct

+ type Scheduler struct

+ func NewScheduler(pool *Pool) *Scheduler

+ func (s *Scheduler) AddTask(ctx context.Context, payload *Task, opts ...SchedulerOption)

+ func (s *Scheduler) PrintAllTasks()

+ func (s *Scheduler) PrintExecutionHistory(taskID string)

+ func (s *Scheduler) RemoveTask(payloadID string)

+ func (s *Scheduler) Start()

+ type SchedulerConfig struct

+ Callback Callback

+ Overlap bool

+ type SchedulerOption func(*ScheduleOptions)

+ func WithInterval(interval time.Duration) SchedulerOption

+ func WithOverlap() SchedulerOption

+ func WithRecurring() SchedulerOption

+ func WithSchedulerCallback(callback Callback) SchedulerOption

+ func WithSchedulerHandler(handler Handler) SchedulerOption

+ type TLSConfig struct

+ CAPath string

+ CertPath string

+ KeyPath string

+ UseTLS bool

+ type Task struct

+ CreatedAt time.Time

+ Error error

+ Expiry time.Time

+ ID string

+ Payload json.RawMessage

+ ProcessedAt time.Time

+ Status string

+ Topic string

+ func NewTask(id string, payload json.RawMessage, nodeKey string) *Task

+ type TaskStorage interface

+ CleanupExpiredTasks func() error

+ DeleteTask func(taskID string) error

+ FetchNextTask func() (*QueueTask, error)

+ GetAllTasks func() ([]*QueueTask, error)

+ GetTask func(taskID string) (*QueueTask, error)

+ SaveTask func(task *QueueTask) error

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL