scheduler

README

Scheduler Package

The scheduler package is a Go library designed for periodic recurring job scheduling and execution using a pool of workers. The package provides an easy-to-use API to manage job lifecycles, handle concurrent execution, and stop jobs gracefully when required.

---

Features

• Job Scheduling: Schedule jobs to run at specified intervals.
• Worker Pool Management: Efficiently manage multiple workers to execute jobs concurrently.
• Graceful Shutdown: Ensure all workers stop cleanly without leaving incomplete tasks.
• Thread-Safe Job Management: Uses sync.Map for safe concurrent job access and modification.
• Context-Based Cancellation: Stop all running jobs and workers using Go's context package.

Documentation

Overview

Package scheduler provides a framework for managing and executing jobs periodically with configurable options for interval, maximum executions, and panic recovery. It supports concurrent job execution using worker goroutines and provides functionality to start, stop, and retrieve jobs by ID.

Index

• Variables
• type Job
  • func NewJob(id string, interval time.Duration, run func(*Job) bool, opts ...JobOption) *Job
  • func (j *Job) ID() string
  • func (j *Job) Interval() time.Duration
  • func (j *Job) IsStopped() bool
  • func (j *Job) Stop()
  • func (j *Job) String() string
• type JobOption
  • func WithMaxExecutions(n uint64) JobOption
  • func WithRecoveryHandler(recoveryHandler func(*Job, any)) JobOption
  • func WithStopOnPanic(b bool) JobOption
• type Scheduler
  • func NewScheduler(bufferSize int, workerCount int, opts ...SchedulerOption) *Scheduler
  • func (s *Scheduler) AddJob(job *Job) error
  • func (s *Scheduler) GetJob(id string) *Job
  • func (s *Scheduler) JobIDs() []string
  • func (s *Scheduler) Stop()
  • func (s *Scheduler) StopJob(jobID string) error
• type SchedulerOption
  • func WithLogger(l *slog.Logger) SchedulerOption
• type Workers

Variables

var (
	ErrJobIDExists     = errors.New("job ID already exists")
	ErrJobNotFound     = errors.New("job not found")
	ErrJobQueueFull    = errors.New("job queue full")
	ErrNilJob          = errors.New("job is nil")
	ErrWorkersStopping = errors.New("workers are stopping")
)

Types

type Job

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

Job represents a periodic task that executes at specified intervals.

A Job continues to run until one of the following conditions is met:

• It is explicitly stopped using Stop().
• It reaches its maximum execution count, set with WithMaxExecutions().
• Its run function returns false, indicating it should not be rescheduled.
• A panic occurs during execution and job created WithStopOnPanic(true).

A RecoveryHandler can be provided via WithRecoveryHandler() to handle panics that occur during the execution of the run function. The default Recovery Handler logs an error.

Users can configure the Job to stop running after a panic using the WithStopOnPanic(). The default behavior is to continue running after a panic.

func NewJob

func NewJob(id string, interval time.Duration, run func(*Job) bool, opts ...JobOption) *Job

NewJob creates and returns a pointer to a job that runs periodically when addded to a Scheduler. The job includes an ID, that must be unique within each Scheduler, a positive interval between executions, and a `run` function that executes at each interval. Optional settings can be applied using JobOptions.

A Job continues to run until one of the following conditions is met:

• It is explicitly stopped using Stop().
• It reaches its maximum execution count, set with WithMaxExecutions().
• Its run function returns false, indicating it should not be rescheduled.
• A panic occurs during execution and job created WithStopOnPanic(true).

If the interval is non-positive or `run` is nil, NewJob panics.

func (*Job) ID

func (j *Job) ID() string

ID returns the unique identifier for the Job, which can be used to reference or manage the job within a scheduler.

func (*Job) Interval

func (j *Job) Interval() time.Duration

Interval returns the time duration between each execution of the Job.

func (*Job) IsStopped

func (j *Job) IsStopped() bool

IsStopped returns true if the Job will no longer be scheduled for execution.

func (*Job) Stop

func (j *Job) Stop()

Stop prevents the job from being re-queued.

func (*Job) String

func (j *Job) String() string

String returns a human-readable representation of the Job.

type JobOption

type JobOption func(*Job)

JobOption configures optional Job parameters.

func WithMaxExecutions

func WithMaxExecutions(n uint64) JobOption

WithMaxExecutions limits the number of times a job will run. If `n` is 0, the job runs indefinitely. Use this option to control how many times the job executes before stopping automatically.

func WithRecoveryHandler

func WithRecoveryHandler(recoveryHandler func(*Job, any)) JobOption

WithRecoveryHandler configures a custom function to handle panics during the job's execution. If the job's `run` function panics, this recovery function is called with the Job and the panic value, enabling custom error handling or cleanup.

func WithStopOnPanic

func WithStopOnPanic(b bool) JobOption

WithStopOnPanic configures the Job to stop running after a panic occurs in the run function. If set to true, the Job will not be rescheduled after a panic. The default behavior is to continue running after a panic.

type Scheduler

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

Scheduler manages job scheduling and execution using worker goroutines.

func NewScheduler

func NewScheduler(bufferSize int, workerCount int, opts ...SchedulerOption) *Scheduler

NewScheduler creates a Scheduler with a job queue buffer and a specified number of worker goroutines.

bufferSize controls the maximum number of jobs that can be queued at once. If the queue is full, new jobs may be rejected until space becomes available.

workerCount sets the number of goroutines that process jobs concurrently. More workers allow multiple jobs to run at the same time.

opts like WithLogger allow further customization.

func (*Scheduler) AddJob

func (s *Scheduler) AddJob(job *Job) error

AddJob adds a job to the scheduler for execution. It returns an error if the job ID already exists.

func (*Scheduler) GetJob

func (s *Scheduler) GetJob(id string) *Job

GetJob retrieves a job by its ID.

func (*Scheduler) JobIDs

func (s *Scheduler) JobIDs() []string

JobIDs returns the IDs of the scheduled jobs.

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop halts all the workers.

func (*Scheduler) StopJob

func (s *Scheduler) StopJob(jobID string) error

StopJob stops a specific job from being re-queued.

type SchedulerOption

type SchedulerOption func(*Scheduler)

SchedulerOption configures optional settings for a Scheduler.

func WithLogger

func WithLogger(l *slog.Logger) SchedulerOption

WithLogger sets a custom logger for the Scheduler.

type Workers

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

Workers manages the execution of jobs using goroutines. It handles job queuing, cancellation, and synchronization.