tasks

Documentation

Index

• Variables
• type BaseScheduler
  • func NewScheduler(threads int, storage Storage, getCurrentTime CurrentTimeResolver, ...) BaseScheduler
• type Callback
• type CurrentTimeResolver
• type Scheduler
  • func (rcv *Scheduler) AddTask(ctx context.Context, task *Task, at time.Time) error
  • func (rcv *Scheduler) ProcessTasks(ctx context.Context, callback Callback) error
  • func (rcv *Scheduler) RemoveTask(ctx context.Context, id string) error
• type Storage
• type Task
• type Worker
  • func NewWorker(callback Callback, storage Storage, getCurrentTime CurrentTimeResolver, ...) *Worker
  • func (rcv *Worker) Do(ctx context.Context, ch <-chan *Task)

Variables

var (
	ErrAddTaskFailed    = errors.New("add task failed")
	ErrRemoveTaskFailed = errors.New("remove task failed")
	ErrGetTaskFailed    = errors.New("get task failed")
	ErrNoTasks          = errors.New("no tasks")
)

var DefaultCurrentTimeResolver = func() time.Time { return time.Now().UTC() }

Types

type BaseScheduler

type BaseScheduler interface {
	AddTask(ctx context.Context, task *Task, at time.Time) error
	ProcessTasks(ctx context.Context, callback Callback) error
	RemoveTask(ctx context.Context, id string) error
}

func NewScheduler

func NewScheduler(
	threads int,
	storage Storage,
	getCurrentTime CurrentTimeResolver,
	logger *log.Logger,
) BaseScheduler

type Callback

type Callback func(ctx context.Context, task *Task) error

type CurrentTimeResolver

type CurrentTimeResolver func() time.Time

type Scheduler

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

func (*Scheduler) AddTask

func (rcv *Scheduler) AddTask(ctx context.Context, task *Task, at time.Time) error

func (*Scheduler) ProcessTasks

func (rcv *Scheduler) ProcessTasks(ctx context.Context, callback Callback) error

func (*Scheduler) RemoveTask

func (rcv *Scheduler) RemoveTask(ctx context.Context, id string) error

type Storage

type Storage interface {
	Add(ctx context.Context, Task *Task, at time.Time) error
	GetNext(ctx context.Context) (*Task, error)
	RemoveBackup(ctx context.Context, id string) error
	GetBack(ctx context.Context, task *Task) error
	RestoreFromBackup(ctx context.Context, earlyThen time.Time) error
	Remove(ctx context.Context, id string) error
}

type Task

type Task struct {
	ID      string
	Time    time.Time
	Delay   time.Duration // if delay > 0 new task time will be based on current time + delay.
	Payload []byte
}

type Worker

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

func NewWorker

func NewWorker(
	callback Callback,
	storage Storage,
	getCurrentTime CurrentTimeResolver,
	logger *log.Logger,
) *Worker

func (*Worker) Do

func (rcv *Worker) Do(ctx context.Context, ch <-chan *Task)