scheduler

Documentation

Overview

Package scheduler fires flows on cron schedules and one-shot times (backlog AO-028) — the basis for "assign and walk away" and recurring digests.

Index

• func NextFire(definition Definition, after time.Time) (time.Time, error)
• func OccurrenceID(scheduleID string, scheduledFor time.Time) string
• type Definition
• type DurableScheduler
  • func NewDurableScheduler(repository Repository, trigger Trigger, displayLocation *time.Location) *DurableScheduler
  • func (s *DurableScheduler) Create(ctx context.Context, definition Definition) (Definition, error)
  • func (s *DurableScheduler) Definitions(context.Context) ([]Definition, error)
  • func (s *DurableScheduler) Start(ctx context.Context) error
  • func (s *DurableScheduler) Stop()
• type Kind
• type Occurrence
  • func OccurrencesForDay(definition Definition, day time.Time, viewerLocation *time.Location) ([]Occurrence, error)
• type OccurrenceState
• type Repository
• type Scheduler
  • func New() *Scheduler
  • func (s *Scheduler) Cron(spec string, fn func()) (cron.EntryID, error)
  • func (s *Scheduler) CronScheduled(spec string, fn func(time.Time)) (cron.EntryID, error)
  • func (s *Scheduler) Once(d time.Duration, fn func()) error
  • func (s *Scheduler) OnceAt(at time.Time, fn func()) error
  • func (s *Scheduler) Start()
  • func (s *Scheduler) Stop()
• type Trigger

Functions

func NextFire

func NextFire(definition Definition, after time.Time) (time.Time, error)

func OccurrenceID

func OccurrenceID(scheduleID string, scheduledFor time.Time) string

Types

type Definition

type Definition struct {
	ID         string    `json:"id"`
	Title      string    `json:"title"`
	Kind       Kind      `json:"kind"`
	Expression string    `json:"expression"`
	FlowID     string    `json:"flow_id"`
	Timezone   string    `json:"timezone"`
	Enabled    bool      `json:"enabled"`
	NextFireAt time.Time `json:"next_fire_at,omitempty"`
	LastFireAt time.Time `json:"last_fire_at,omitempty"`
	CreatedAt  time.Time `json:"created_at"`
	UpdatedAt  time.Time `json:"updated_at"`
}

type DurableScheduler

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

func NewDurableScheduler

func NewDurableScheduler(repository Repository, trigger Trigger, displayLocation *time.Location) *DurableScheduler

func (*DurableScheduler) Create

func (s *DurableScheduler) Create(ctx context.Context, definition Definition) (Definition, error)

Create persists and hot-registers a definition in the running daemon.

func (*DurableScheduler) Definitions

func (s *DurableScheduler) Definitions(context.Context) ([]Definition, error)

func (*DurableScheduler) Start

func (s *DurableScheduler) Start(ctx context.Context) error

Start reloads durable definitions before the in-process clock begins.

func (*DurableScheduler) Stop

func (s *DurableScheduler) Stop()

Stop joins active callbacks and permanently closes this manager. A daemon restart constructs a fresh manager from the durable repository.

type Kind

type Kind string

const (
	KindCron Kind = "cron"
	KindOnce Kind = "once"
)

type Occurrence

type Occurrence struct {
	ID           string          `json:"id"`
	ScheduleID   string          `json:"schedule_id"`
	RunID        string          `json:"run_id,omitempty"`
	ScheduledFor time.Time       `json:"scheduled_for"`
	State        OccurrenceState `json:"state"`
	Error        string          `json:"error,omitempty"`
	CreatedAt    time.Time       `json:"created_at"`
	UpdatedAt    time.Time       `json:"updated_at"`
}

func OccurrencesForDay

func OccurrencesForDay(definition Definition, day time.Time, viewerLocation *time.Location) ([]Occurrence, error)

OccurrencesForDay expands one durable definition into real occurrences whose instants fall inside the requested viewer-local calendar day.

type OccurrenceState

type OccurrenceState string

const (
	OccurrenceScheduled OccurrenceState = "scheduled"
	OccurrenceFired     OccurrenceState = "fired"
	OccurrenceRunning   OccurrenceState = "running"
	OccurrenceSucceeded OccurrenceState = "succeeded"
	OccurrenceFailed    OccurrenceState = "failed"
	OccurrenceCanceled  OccurrenceState = "canceled"
)

type Repository

type Repository interface {
	CreateSchedule(Definition) error
	ListSchedules() ([]Definition, error)
	UpsertScheduleOccurrence(Occurrence) error
	TransitionScheduleOccurrence(id string, from, to OccurrenceState, runID, errorMessage string) (bool, error)
	UpdateScheduleFire(id string, lastFireAt, nextFireAt time.Time) error
}

type Scheduler

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

Scheduler runs cron and one-shot triggers.

func New

func New() *Scheduler

New builds a scheduler with seconds-precision cron support.

func (*Scheduler) Cron

func (s *Scheduler) Cron(spec string, fn func()) (cron.EntryID, error)

Cron registers a recurring trigger. spec is a 6-field cron (with seconds), e.g. "0 0 9 * * *" for 09:00 daily, or "@every 1h".

func (*Scheduler) CronScheduled

func (s *Scheduler) CronScheduled(spec string, fn func(time.Time)) (cron.EntryID, error)

CronScheduled registers a recurring trigger and reports the occurrence time planned by the cron clock, even when its callback starts late.

func (*Scheduler) Once

func (s *Scheduler) Once(d time.Duration, fn func()) error

Once registers a one-shot trigger firing after d.

func (*Scheduler) OnceAt

func (s *Scheduler) OnceAt(at time.Time, fn func()) error

OnceAt registers a one-shot trigger firing at the given time.

func (*Scheduler) Start

func (s *Scheduler) Start()

Start begins firing cron triggers.

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop halts cron triggers and cancels pending one-shots.

type Trigger

type Trigger func(context.Context, Definition, Occurrence) error