queue

Feb 4, 2026

Changes in this version

+ var DefaultBlockingTimeout = time.Second

+ var DefaultCallbackWaitInterval = 10 * time.Millisecond

+ var DefaultNetworkRetryDelay = 100 * time.Millisecond

type BaseQueue

+ func (q *BaseQueue[T]) AddInflight() bool

+ func (q *BaseQueue[T]) DoneInflight()

+ func (q *BaseQueue[T]) GracefulClose()

+ func (q *BaseQueue[T]) IsClosing() bool

+ func (q *BaseQueue[T]) ShouldSendToDLQ(ctx context.Context, msg Message[T]) (bool, error)

+ func (q *BaseQueue[T]) UnpackMessage(packedData []byte) (Message[T], error)

+ type MsgpackMessage struct

+ func NewMsgpackMessage[T any](data T) *MsgpackMessage[T]

+ func (m *MsgpackMessage[T]) AddRetryCount()

+ func (m *MsgpackMessage[T]) CreatedAt() time.Time

+ func (m *MsgpackMessage[T]) RefreshRetryCount()

+ func (m *MsgpackMessage[T]) RefreshUpdatedAt()

+ func (m *MsgpackMessage[T]) RetryCount() int

+ func (m *MsgpackMessage[T]) TotalRetryCount() int

+ func (m *MsgpackMessage[T]) UpdatedAt() time.Time

Jan 30, 2026

Changes in this version

type UnifiedFactory

+ func (f *UnifiedFactory) CacheSize() int

+ func (f *UnifiedFactory) ClearCache()

+ func (f *UnifiedFactory) RemoveFromCache(name string) bool

Jan 30, 2026

Changes in this version

+ const QueueInfoTypeDLQ

+ const QueueInfoTypeMain

+ const QueueInfoTypeRetry

+ var DefaultRetryQueueCapacity = 10000

+ var DefaultUnlimitedCapacity = 1000000

+ func IsQueueTypeRegistered(queueType QueueType) bool

+ func IsQueueTypeSupported(queueType QueueType) bool

+ func RegisterQueueCreator[T any](queueType QueueType, creator QueueCreator[T])

+ func RegisterQueueValidator(queueType QueueType, validator QueueConfigValidator)

+ func ValidateUnifiedConfig(config UnifiedQueueConfig) error

+ func WithRetryQueueCapacity(capacity int) func(*Config)

+ func WithUnlimitedCapacity(capacity int) func(*Config)

type BaseQueue

+ func (q *BaseQueue[T]) ExitChannel() <-chan struct{}

+ func (q *BaseQueue[T]) GetCallbacks() []Handler[T]

+ func (q *BaseQueue[T]) GetRetryQueueKey() string

+ func (q *BaseQueue[T]) GetRetryQueueName() string

+ func (q *BaseQueue[T]) HasCallbacks() bool

+ func (q *BaseQueue[T]) IsClosed() bool

type Config

+ RetryQueueCapacity int

+ UnlimitedCapacity int

+ type Discoverable interface

+ DiscoverAllQueues func(ctx context.Context, pattern string) ([]QueueInfo, error)

+ DiscoverQueues func(ctx context.Context, pattern string) ([]QueueInfo, error)

type MemoryQueue

+ func (q *MemoryQueue[T]) EnqueueToRetryQueue(ctx context.Context, data T) error

+ func (q *MemoryQueue[T]) Stats() QueueStats

+ type MemoryQueueConfig struct

+ type MemoryQueueFactory struct

+ func NewMemoryQueueFactory() (*MemoryQueueFactory, error)

+ func (f *MemoryQueueFactory) DiscoverAllQueues(ctx context.Context, pattern string) ([]QueueInfo, error)

+ func (f *MemoryQueueFactory) DiscoverQueues(ctx context.Context, pattern string) ([]QueueInfo, error)

type Queue

+ func GetOrCreate[T any](f *UnifiedFactory, name string, defaultMsg Message[T], options ...Option) (Queue[T], error)

+ func MemoryGetOrCreate[T any](f *MemoryQueueFactory, name string, defaultMsg Message[T], options ...Option) (Queue[T], error)

+ func RedisGetOrCreate[T any](f *RedisFactory, name string, defaultMsg Message[T], options ...Option) (Queue[T], error)

+ type QueueConfigValidator interface

+ Validate func(rawConfig json.RawMessage) error

+ type QueueCreator func(ctx context.Context, name string, defaultMsg Message[T], ...) (Queue[T], error)

+ type QueueInfo struct

+ Depth int64

+ Key string

+ Name string

+ Type string

+ type QueueStats struct

+ Capacity int

+ ConsumerCount int

+ Depth int64

+ Inflight int64

+ RetryCapacity int

+ RetryDepth int64

+ type QueueType string

+ const QueueTypeMemory

+ const QueueTypeRedis

+ func GetRegisteredQueueTypes() []QueueType

+ type RedisFactory struct

+ func NewRedisFactory(redisClient redis.Cmdable) (*RedisFactory, error)

+ func (f *RedisFactory) Client() redis.Cmdable

+ func (f *RedisFactory) DiscoverAllQueues(ctx context.Context, pattern string) ([]QueueInfo, error)

+ func (f *RedisFactory) DiscoverQueues(ctx context.Context, pattern string) ([]QueueInfo, error)

type RedisQueue

+ func (q *RedisQueue[T]) Close()

+ func (q *RedisQueue[T]) EnqueueToRetryQueue(ctx context.Context, data T) error

+ func (q *RedisQueue[T]) Stats() QueueStats

+ type RedisQueueConfig struct

+ Addr string

+ DB int

+ Password string

+ type RetryQueueEnqueuer interface

+ EnqueueToRetryQueue func(ctx context.Context, data T) error

type SafeQueue

+ func GetOrCreateSafe[T any](f *UnifiedFactory, name string, defaultMsg Message[T], options ...Option) (SafeQueue[T], error)

+ func MemoryGetOrCreateSafe[T any](f *MemoryQueueFactory, name string, defaultMsg Message[T], options ...Option) (SafeQueue[T], error)

+ func RedisGetOrCreateSafe[T any](f *RedisFactory, name string, defaultMsg Message[T], options ...Option) (SafeQueue[T], error)

type SimpleQueue

+ func (q *SimpleQueue[T]) IsStatsProvider() bool

+ func (q *SimpleQueue[T]) Stats() QueueStats

+ type StatsProvider interface

+ Stats func() QueueStats

+ type TypedFactory struct

+ func NewTypedFactory[T any](config UnifiedQueueConfig, defaultMsg Message[T]) (*TypedFactory[T], error)

+ func (f *TypedFactory[T]) Config() UnifiedQueueConfig

+ func (f *TypedFactory[T]) Factory() *UnifiedFactory

+ func (f *TypedFactory[T]) GetOrCreate(name string, options ...Option) (Queue[T], error)

+ func (f *TypedFactory[T]) GetOrCreateSafe(name string, options ...Option) (SafeQueue[T], error)

+ type UnifiedFactory struct

+ func NewUnifiedFactory(config UnifiedQueueConfig) (*UnifiedFactory, error)

+ func (f *UnifiedFactory) Config() UnifiedQueueConfig

+ func (f *UnifiedFactory) DiscoverAllQueues(ctx context.Context, pattern string) ([]QueueInfo, error)

+ func (f *UnifiedFactory) DiscoverQueues(ctx context.Context, pattern string) ([]QueueInfo, error)

+ type UnifiedQueueConfig struct

+ BackendConfig json.RawMessage

+ CallbackParallel bool

+ ConsumerCount int

+ MaxHandleFailures int

+ MaxSize int

+ RedisClient redis.Cmdable

+ RetryQueueCapacity int

+ Type QueueType

+ UnlimitedCapacity int

Jan 4, 2026

Changes in this version

+ func WithCallbackParallelExecution(enable bool) func(*Config)

+ func WithCallbackTimeout(timeout time.Duration) func(*Config)

type Config

+ CallbackParallelExecution bool

+ CallbackTimeout time.Duration

+ type SimpleQueueConfig struct

+ Logger *zerolog.Logger

+ type SimpleQueueOption func(*SimpleQueueConfig)

+ func WithLogger(logger *zerolog.Logger) SimpleQueueOption

May 30, 2025

Changes in this version

+ var ErrInvalidData = errors.New("invalid data")

type BaseQueue

+ func (q *BaseQueue[T]) BDequeue(ctx context.Context) (Message[T], error)

+ func (q *BaseQueue[T]) BEnqueue(ctx context.Context, data T) error

type MemoryQueue

+ func (q *MemoryQueue[T]) BDequeue(ctx context.Context) (Message[T], error)

+ func (q *MemoryQueue[T]) BEnqueue(ctx context.Context, data T) error

type Queue

+ BDequeue func(context.Context) (Message[T], error)

+ BEnqueue func(ctx context.Context, data T) error

type RedisQueue

+ func (q *RedisQueue[T]) BDequeue(ctx context.Context) (Message[T], error)

+ func (q *RedisQueue[T]) BEnqueue(ctx context.Context, data T) error

type SimpleQueue

+ func (q *SimpleQueue[T]) BDequeue(ctx context.Context) (Message[T], error)

+ func (q *SimpleQueue[T]) BEnqueue(ctx context.Context, data T) error

May 28, 2025

Changes in this version

+ type BaseDLQ struct

+ func (q *BaseDLQ[T]) AssociatedQueue() Queue[T]

+ func (q *BaseDLQ[T]) Redrive(ctx context.Context, items int) error

type BaseQueue

+ func (q *BaseQueue[T]) DLQ() (DLQ[T], error)

+ func (q *BaseQueue[T]) GetDeadletterQueueKey() string

+ func (q *BaseQueue[T]) GetDeadletterQueueName() string

+ func (q *BaseQueue[T]) SetDLQ(dlq DLQ[T])

type RedisQueue

+ func (q *RedisQueue[T]) Purge(ctx context.Context) error

May 28, 2025

Changes in this version

+ const Namespace

type BaseQueue

+ func (q *BaseQueue[T]) GetConfig() *Config

+ func (q *BaseQueue[T]) GetQueueKey() string

+ func (q *BaseQueue[T]) TriggerCallbacks(msg Message[T])

May 14, 2025

Changes in this version

type BaseQueue

+ func (q *BaseQueue) GetLocker() sync.Locker

type Config

+ LockerGenerator locker.SyncLockerGenerator

May 11, 2025

Changes in this version

+ const UnlimitedSize

type Factory

+ GetOrCreateSafe func(name string, options ...Option) (SafeQueue, error)

type MemoryFactory

+ func (f *MemoryFactory) GetOrCreateSafe(name string, options ...Option) (SafeQueue, error)

type RedisQueueFactory

+ func (f *RedisQueueFactory) GetOrCreateSafe(name string, options ...Option) (SafeQueue, error)

type SimpleQueue

+ func NewSimpleQueue(queue Queue) (*SimpleQueue, error)

Apr 6, 2025

Changes in this version

+ func WithConsumerCount(consumerCount int) func(*Config)

+ func WithMaxHandleFailures(maxHandleFailures int) func(*Config)

+ func WithMaxRetries(maxRetries int) func(*Config)

+ func WithMaxSize(maxSize int) func(*Config)

+ func WithMessage(message Message) func(*Config)

+ func WithMessageIDGenerator(generator message.MessageIDGenerator) func(*Config)

+ func WithPollInterval(pollInterval time.Duration) func(*Config)

type BaseQueue

+ func (q *BaseQueue) MaxHandleFailures() int

+ func (q *BaseQueue) NewMessage(data []byte) (Message, error)

+ func (q *BaseQueue) Pack(data []byte) ([]byte, error)

+ func (q *BaseQueue) Unpack(data []byte) (Message, error)

+ func (q *BaseQueue) ValidateQueueClosed() error

+ type Config struct

+ ConsumerCount int

+ MaxHandleFailures int

+ MaxRetries int

+ MaxSize int

+ Message Message

+ MessageIDGenerator message.MessageIDGenerator

+ PollInterval time.Duration

+ type DLQ interface

+ AssociatedQueue func() Queue

+ Redrive func(items int) error

+ type DLQer interface

+ DLQ func() (DLQ, error)

+ type JsonMessage struct

+ func NewJsonMessage() *JsonMessage

+ func (m *JsonMessage) AddRetryCount()

+ func (m *JsonMessage) CreatedAt() time.Time

+ func (m *JsonMessage) RefreshRetryCount()

+ func (m *JsonMessage) RefreshUpdatedAt()

+ func (m *JsonMessage) RetryCount() int

+ func (m *JsonMessage) TotalRetryCount() int

+ func (m *JsonMessage) UpdatedAt() time.Time

type MemoryQueue

+ func (q *MemoryQueue) Purge() error

+ type Message interface

+ AddRetryCount func()

+ CreatedAt func() time.Time

+ RefreshRetryCount func()

+ RefreshUpdatedAt func()

+ RetryCount func() int

+ TotalRetryCount func() int

+ UpdatedAt func() time.Time

+ type Option func(*Config)

+ type Purgeable interface

+ Purge func() error

type Queue

+ MaxHandleFailures func() int

type SafeQueue

+ IsDLQSupported func() bool

+ IsPurgeable func() bool

+ type SimpleQueue struct

+ func (q *SimpleQueue) Close()

+ func (q *SimpleQueue) DLQ() (DLQ, error)

+ func (q *SimpleQueue) Dequeue() (Message, error)

+ func (q *SimpleQueue) Enqueue(data []byte) error

+ func (q *SimpleQueue) IsDLQSupported() bool

+ func (q *SimpleQueue) IsPurgeable() bool

+ func (q *SimpleQueue) IsRecoverable() bool

+ func (q *SimpleQueue) Kind() Kind

+ func (q *SimpleQueue) MaxHandleFailures() int

+ func (q *SimpleQueue) MaxSize() int

+ func (q *SimpleQueue) Name() string

+ func (q *SimpleQueue) Purge() error

+ func (q *SimpleQueue) Recover(msg Message) error

+ func (q *SimpleQueue) Subscribe(cb Handler)

+ func (q *SimpleQueue) Unwrap() Queue

Apr 5, 2025

Changes in this version

type QueueOptions

+ ConsumerCount int

Apr 5, 2025

Changes in this version

+ const UnlimitedMaxSize

+ var DefaultMaxRetries = 10

+ var DefaultOptions = QueueOptions

+ var ErrQueueRecovered = errors.New("queue recovered")

+ type BaseQueue struct

+ func NewBaseQueue(name string, options *QueueOptions) *BaseQueue

+ func (q *BaseQueue) Close()

+ func (q *BaseQueue) Dequeue() ([]byte, error)

+ func (q *BaseQueue) Enqueue(data []byte) error

+ func (q *BaseQueue) Kind() Kind

+ func (q *BaseQueue) MaxSize() int

+ func (q *BaseQueue) Name() string

+ func (q *BaseQueue) Subscribe(cb Handler)

type MemoryQueue

+ func (q *MemoryQueue) Name() string

type Queue

+ Close func()

+ Name func() string

+ type QueueOptions struct

+ MaxRetries int

+ MaxSize int

+ PollInterval time.Duration

+ type Recoverable interface

+ Recover func([]byte) error

+ type RedisQueue struct

+ func NewRedisQueue(conn rmq.Connection, name string, options *QueueOptions) (*RedisQueue, error)

+ func (q *RedisQueue) Dequeue() ([]byte, error)

+ func (q *RedisQueue) Enqueue(data []byte) error

+ func (q *RedisQueue) MaxSize() int

+ func (q *RedisQueue) Recover(b []byte) error

+ func (q *RedisQueue) Subscribe(cb Handler)

+ type RedisQueueFactory struct

+ func NewRedisQueueFactory(redisClient redis.Cmdable) *RedisQueueFactory

+ func (f *RedisQueueFactory) GetOrCreate(name string, options ...func(*QueueOptions)) (Queue, error)

type SafeQueue

+ func (q *SafeQueue) Close()

+ func (q *SafeQueue) Name() string

Apr 5, 2025

Changes in this version

+ var DefaultPollInterval = 20 * time.Millisecond

+ var ErrQueueClosed = errors.New("queue is closed")

+ var ErrQueueEmpty = errors.New("queue is empty")

+ var ErrQueueFull = errors.New("queue is full")

+ type Factory interface

+ GetOrCreate func(name string, maxSize int) Queue

+ type Handler func([]byte) error

+ type Kind uint8

+ const KindFIFO

+ const KindStandard

+ type MemoryFactory struct

+ func NewMemoryFactory() *MemoryFactory

+ func (f *MemoryFactory) GetOrCreate(name string, maxSize int) Queue

+ type MemoryQueue struct

+ func NewMemoryQueue(maxSize int, pollInterval time.Duration) *MemoryQueue

+ func (q *MemoryQueue) Close()

+ func (q *MemoryQueue) Dequeue() ([]byte, error)

+ func (q *MemoryQueue) Enqueue(data []byte) error

+ func (q *MemoryQueue) Kind() Kind

+ func (q *MemoryQueue) MaxSize() int

+ func (q *MemoryQueue) Recover(b []byte) error

+ func (q *MemoryQueue) Subscribe(cb Handler)

+ type Queue interface

+ Dequeue func() ([]byte, error)

+ Enqueue func([]byte) error

+ Kind func() Kind

+ MaxSize func() int

+ Subscribe func(h Handler)

+ type RecoverableQueue interface

+ Recover func([]byte) error

+ type SafeQueue struct

+ func NewSafeQueue(queue Queue) *SafeQueue

+ func (q *SafeQueue) Dequeue() ([]byte, error)

+ func (q *SafeQueue) Enqueue(data []byte) error

+ func (q *SafeQueue) IsRecoverable() bool

+ func (q *SafeQueue) Kind() Kind

+ func (q *SafeQueue) MaxSize() int

+ func (q *SafeQueue) Recover(b []byte) error

+ func (q *SafeQueue) Subscribe(cb Handler)