scheduler

Documentation

Index

• func ValidateItem(item Item) error
• func ValidateSchedule(schedule Schedule) error
• type Derivation
• type DerivationFunc
  • func (f DerivationFunc) Derive(ctx context.Context, tx *sql.Tx, plans []que.Plan) ([]que.Plan, error)
• type FileProvider
  • func (fp *FileProvider) Provide() (Schedule, error)
• type Item
• type MemProvider
  • func (mp *MemProvider) Provide() (Schedule, error)
• type Provider
• type RecoveryPolicy
• type Schedule
  • func Provide(r io.Reader) (Schedule, error)
• type Scheduler
  • func (sc *Scheduler) Perform(ctx context.Context, job que.Job) error
  • func (sc *Scheduler) Prepare(ctx context.Context) error

Functions

func ValidateItem

func ValidateItem(item Item) error

ValidateItem validates item is legal.

func ValidateSchedule

func ValidateSchedule(schedule Schedule) error

ValidateSchedule validates schedule content is legal.

Types

type Derivation

type Derivation interface {
	Derive(context.Context, *sql.Tx, []que.Plan) ([]que.Plan, error)
}

Derivation derives new plans from scheduled plans. Derived plans will replace scheduled plans.

type DerivationFunc

type DerivationFunc func(context.Context, *sql.Tx, []que.Plan) ([]que.Plan, error)

DerivationFunc wraps a function as Derivation.

func (DerivationFunc) Derive

func (f DerivationFunc) Derive(ctx context.Context, tx *sql.Tx, plans []que.Plan) ([]que.Plan, error)

Derive implements Derive of Derivation.

type FileProvider

type FileProvider struct {
	Filename string
}

FileProvider provides a schedule from a file.

func (*FileProvider) Provide

func (fp *FileProvider) Provide() (Schedule, error)

type Item

type Item struct {
	Queue string `yaml:"queue"`
	// Args is a json array.
	Args string `yaml:"args"`
	// See https://en.wikipedia.org/wiki/Cron or https://github.com/robfig/cron
	Cron string `yaml:"cron"`

	RecoveryPolicy RecoveryPolicy `yaml:"recoveryPolicy"`

	RetryPolicy que.RetryPolicy `yaml:"retryPolicy"`
}

Item represents a job need to enqueue queue according to cron.

type MemProvider

type MemProvider struct {
	Schedule Schedule
}

MemProvider provides a schedule from given Schedule.

func (*MemProvider) Provide

func (mp *MemProvider) Provide() (Schedule, error)

type Provider

type Provider interface {
	Provide() (Schedule, error)
}

Provider provides a schedule.

type RecoveryPolicy

type RecoveryPolicy string

RecoveryPolicy guides how to process schedule after a long period of downtime.

const (
	// Ignore ignores missed events and schedule each item from now.
	Ignore RecoveryPolicy = "ignore"
	// Reparation schedule each item from last scheduled item run time to now.
	Reparation RecoveryPolicy = "reparation"
)

type Schedule

type Schedule map[string]Item

Schedule is a set of named items.

func Provide

func Provide(r io.Reader) (Schedule, error)

Provide obtains a schedule from a r.

type Scheduler

type Scheduler struct {
	DB *sql.DB
	// Queue is used by scheduler ennqueue self.
	Queue string
	// Enqueue saves a set of jobs according to plans.
	Enqueue func(context.Context, *sql.Tx, ...que.Plan) ([]int64, error)
	// Provider privide a list of named schedule.
	// Scheduler enqueues new plans according to privided schedule.
	Provider Provider
	// Derivations is list of Derivation.
	// The key of provided Schedule corresponding to the key of Derivations.
	Derivations map[string]Derivation
}

Scheduler enqueue jobs according to provided schedule. Prepare must be called in application initializtion. Scheduler with same Queue can run in different machines.

func (*Scheduler) Perform

func (sc *Scheduler) Perform(ctx context.Context, job que.Job) error

Perform is Perform function of que.WorkerOptions.

func (*Scheduler) Prepare

func (sc *Scheduler) Prepare(ctx context.Context) error

Prepare auto schedules self in given queue at first. Prepare must be called in application initializtion.