schedule

Documentation

Index

• func ClearRejectedSchedulesCache()
• func IsScheduleRejected(schedule string) bool
• type Config
  • func DefaultConfig() Config
• type Executor
• type Heap
  • func NewHeap[T Item]() *Heap[T]
  • func (h *Heap[T]) Get(id string) (T, bool)
  • func (h *Heap[T]) Len() int
  • func (h *Heap[T]) Peek() (T, bool)
  • func (h *Heap[T]) Pop() (T, bool)
  • func (h *Heap[T]) PopIfDue() (T, bool)
  • func (h *Heap[T]) Push(item T)
  • func (h *Heap[T]) Remove(id string) bool
• type Item
• type Scheduler
  • func NewScheduler[T Item](executor Executor[T], config Config) *Scheduler[T]
  • func (s *Scheduler[T]) Add(item T) error
  • func (s *Scheduler[T]) Count() int
  • func (s *Scheduler[T]) Get(id string) (T, bool)
  • func (s *Scheduler[T]) Remove(id string)
  • func (s *Scheduler[T]) Run(ctx context.Context) error
  • func (s *Scheduler[T]) SetIsLeader(fn func() bool)
  • func (s *Scheduler[T]) SetItemResolver(fn func(T) T)
  • func (s *Scheduler[T]) ValidateSchedule(schedule string) error

Functions

func ClearRejectedSchedulesCache

func ClearRejectedSchedulesCache()

ClearRejectedSchedulesCache clears the cache of rejected schedules. This is primarily useful for testing.

func IsScheduleRejected

func IsScheduleRejected(schedule string) bool

IsScheduleRejected returns true if the given schedule has been cached as rejected. This is primarily useful for testing.

Types

type Config

type Config struct {
	// MissedWindow is how late an item can be before skipping execution (default: 5m)
	MissedWindow time.Duration
	// IdleInterval is how long to sleep when heap is empty (default: 1m)
	IdleInterval time.Duration
}

Config configures the generic scheduler behavior.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns the default scheduler configuration.

type Executor

type Executor[T Item] func(ctx context.Context, item T) error

Executor is called when a scheduled item is due for execution. The executor should NOT reschedule the item - that's handled by the scheduler.

type Heap

type Heap[T Item] struct {
	// contains filtered or unexported fields
}

Heap is a generic min-heap ordered by NextExecution time. It is safe for concurrent use.

func NewHeap

func NewHeap[T Item]() *Heap[T]

NewHeap creates a new empty heap.

func (*Heap[T]) Get

func (h *Heap[T]) Get(id string) (T, bool)

Get returns the item with the given ID, or the zero value of T and false if not found.

func (*Heap[T]) Len

func (h *Heap[T]) Len() int

Len returns the number of items in the heap.

func (*Heap[T]) Peek

func (h *Heap[T]) Peek() (T, bool)

Peek returns the next item without removing it from the heap. Returns the zero value of T and false if the heap is empty.

func (*Heap[T]) Pop

func (h *Heap[T]) Pop() (T, bool)

Pop removes and returns the next item from the heap. Returns the zero value of T and false if the heap is empty.

func (*Heap[T]) PopIfDue

func (h *Heap[T]) PopIfDue() (T, bool)

PopIfDue removes and returns the next item if it's due (NextExecution <= now). Returns the zero value of T and false if the heap is empty or the next item is not yet due.

func (*Heap[T]) Push

func (h *Heap[T]) Push(item T)

Push adds or updates an item in the heap. If an item with the same ID exists, it updates the entry and fixes the heap position.

func (*Heap[T]) Remove

func (h *Heap[T]) Remove(id string) bool

Remove removes an item by ID. Returns true if the item was found and removed.

type Item

type Item interface {
	// ID returns a unique identifier for this item.
	ID() string
	// Schedule returns the cron expression for this item.
	Schedule() string
	// NextExecution returns when this item should next execute.
	NextExecution() time.Time
	// SetNextExecution updates the next execution time.
	SetNextExecution(time.Time)
	// HeapIndex returns the current index in the heap (-1 if not in heap).
	HeapIndex() int
	// SetHeapIndex updates the heap index.
	SetHeapIndex(int)
}

Item represents anything that can be scheduled. Implementations must be pointer types to support mutation of heap index.

type Scheduler

type Scheduler[T Item] struct {
	// contains filtered or unexported fields
}

Scheduler is a generic cron-based scheduler.

func NewScheduler

func NewScheduler[T Item](executor Executor[T], config Config) *Scheduler[T]

NewScheduler creates a new generic scheduler.

func (*Scheduler[T]) Add

func (s *Scheduler[T]) Add(item T) error

Add adds or updates an item in the schedule. The item's Schedule() must return a valid cron expression.

func (*Scheduler[T]) Count

func (s *Scheduler[T]) Count() int

Count returns the number of scheduled items.

func (*Scheduler[T]) Get

func (s *Scheduler[T]) Get(id string) (T, bool)

Get returns an item by ID, or nil and false if not found.

func (*Scheduler[T]) Remove

func (s *Scheduler[T]) Remove(id string)

Remove removes an item by ID.

func (*Scheduler[T]) Run

func (s *Scheduler[T]) Run(ctx context.Context) error

Run executes the scheduler loop until context is cancelled.

func (*Scheduler[T]) SetIsLeader

func (s *Scheduler[T]) SetIsLeader(fn func() bool)

SetIsLeader sets the leadership check callback. Only the leader actually executes items; non-leaders still maintain the heap.

func (*Scheduler[T]) SetItemResolver

func (s *Scheduler[T]) SetItemResolver(fn func(T) T)

SetItemResolver sets the item resolver callback. When set, the scheduler calls this before rescheduling an item to get the latest version. This helps avoid race conditions where an item is updated while being executed.

func (*Scheduler[T]) ValidateSchedule

func (s *Scheduler[T]) ValidateSchedule(schedule string) error

ValidateSchedule checks if a cron schedule expression is valid. Returns nil if valid, or an error describing why it's invalid. Use this to pre-validate schedules before calling Add().