queue

type HandlerFunc func(ctx context.Context, storage Storage, queueNum int, tasks []Task) (HandlerStats, error)

type HandlerStats struct {
	Processed int
	Errors    int
	Retried   int
	Duration  time.Duration
	LastError error
}

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

type Option func(*MemoryStorage)

type PutTaskRequest struct {
	QueueNum  int
	Data      []byte
	StartTime *time.Time
	Delay     time.Duration
}

type QueueStat struct {
	QueueNum     int
	TotalTasks   int64
	ReadyTasks   int64
	TakenTasks   int64
	DelayedTasks int64
}

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

type RetryInfo struct {
	TaskID        int64
	NextStartTime time.Time
}

type Stats struct {
	QueueStats map[int]QueueStat
}

type Storage interface {
	// GetTask atomically takes up to count tasks from the given queue.
	// Taken tasks are marked as "in progress" and will not be returned
	// to other consumers until the visibility timeout expires.
	GetTask(ctx context.Context, queueNum int, count int) ([]Task, error)

	// DoneTask marks tasks as completed. They are removed from storage.
	DoneTask(ctx context.Context, taskIDs []int64) error

	// RetryTask returns tasks to the queue with the specified next start time.
	// Attempts is incremented, PrevStartTime is set to the current StartTime.
	RetryTask(ctx context.Context, retries []RetryInfo) error

	// PutTask adds a new task to the queue.
	PutTask(ctx context.Context, req PutTaskRequest) error

	// GetStat returns statistics for all queues.
	GetStat(ctx context.Context) (*Stats, error)
}

type Task struct {
	ID            int64
	QueueNum      int
	Data          []byte
	Attempts      int
	CreatedAt     time.Time
	StartTime     time.Time
	PrevStartTime time.Time
}

type WorkerOption func(*QueueWorker)