Documentation
¶
Overview ¶
Package queue provides queue management for the NZB import service.
Index ¶
- func IsDatabaseContentionError(err error) bool
- type Claimer
- type ItemProcessor
- type Manager
- func (m *Manager) CancelProcessing(itemID int64) error
- func (m *Manager) ExecuteItem(ctx context.Context, itemID int64) error
- func (m *Manager) IsPaused() bool
- func (m *Manager) IsRunning() bool
- func (m *Manager) Pause()
- func (m *Manager) Resize(ctx context.Context, newCount int) error
- func (m *Manager) Resume()
- func (m *Manager) Start(ctx context.Context) error
- func (m *Manager) Stop(ctx context.Context) error
- type ManagerConfig
- type QueueEventListener
- type QueueRepository
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func IsDatabaseContentionError ¶
IsDatabaseContentionError checks if an error is a retryable database contention error
Types ¶
type Claimer ¶
type Claimer struct {
// contains filtered or unexported fields
}
Claimer handles claiming queue items with retry logic
func (*Claimer) ClaimWithRetry ¶
func (c *Claimer) ClaimWithRetry(ctx context.Context, workerID int) (*database.ImportQueueItem, error)
ClaimWithRetry attempts to claim a queue item with exponential backoff retry logic
type ItemProcessor ¶
type ItemProcessor interface {
// ProcessItem processes a single queue item and returns the resulting path or an error
ProcessItem(ctx context.Context, item *database.ImportQueueItem) (string, error)
// HandleSuccess handles successful processing
HandleSuccess(ctx context.Context, item *database.ImportQueueItem, resultingPath string) error
// HandleFailure handles failed processing
HandleFailure(ctx context.Context, item *database.ImportQueueItem, err error)
}
ItemProcessor defines the interface for processing queue items
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager manages queue workers and processing
func NewManager ¶
func NewManager(cfg ManagerConfig, repository *database.QueueRepository, processor ItemProcessor, listener QueueEventListener) *Manager
NewManager creates a new queue manager
func (*Manager) CancelProcessing ¶
CancelProcessing cancels processing for a specific item
func (*Manager) ExecuteItem ¶ added in v0.3.0
ExecuteItem manually triggers processing for a specific queue item, bypassing concurrency limits.
func (*Manager) Resize ¶ added in v0.3.0
Resize dynamically adjusts the number of queue workers. When scaling down, excess workers finish their current item before stopping.
type ManagerConfig ¶
type ManagerConfig struct {
Workers int
ConfigGetter config.ConfigGetter
}
ManagerConfig holds configuration for the queue manager
type QueueEventListener ¶
type QueueEventListener interface {
OnItemClaimed(ctx context.Context, item *database.ImportQueueItem)
}
QueueEventListener receives notifications about queue item lifecycle events.
type QueueRepository ¶
type QueueRepository interface {
ClaimNextQueueItem(ctx context.Context) (*database.ImportQueueItem, error)
}
QueueRepository defines the interface for queue database operations
Source Files