cron

README

CRON

Installation

go get -u github.com/gowool/cron

License

Distributed under MIT License, please see license file within the code for more details.

Documentation

Index

• Variables
• func NewRedisLocker(config LockerConfig, client redis.UniversalClient) gocron.Locker
• func NewScheduler(config Config, jobSyncer Syncer, logger *zap.Logger, ...) (gocron.Scheduler, error)
• func NewSchedulerWithRedisLocker(config Config, jobSyncer Syncer, client redis.UniversalClient, ...) (gocron.Scheduler, error)
• type Config
  • func (c *Config) InitDefaults()
• type Job
• type JobSyncer
  • func NewSyncer(provider Provider, resolver Resolver, logger *zap.Logger) *JobSyncer
  • func (syncer JobSyncer) Sync(ctx context.Context, s gocron.Scheduler)
• type JobType
  • func (t JobType) String() string
• type LockerConfig
  • func (c *LockerConfig) InitDefaults()
• type Logger
  • func (l Logger) Debug(msg string, args ...any)
  • func (l Logger) Error(msg string, args ...any)
  • func (l Logger) Info(msg string, args ...any)
  • func (l Logger) Warn(msg string, args ...any)
• type MiddlewareFunc
  • func MetricsMiddleware() (MiddlewareFunc, error)
  • func TracingMiddleware() MiddlewareFunc
• type Provider
• type Resolver
• type Syncer
• type Task
• type TaskFunc
• type TaskResolver
  • func NewTaskResolver(tasks map[JobType]TaskFunc, middleware ...MiddlewareFunc) *TaskResolver
  • func (r *TaskResolver) Resolve(ctx context.Context, job Job) (Task, error)

Variables

var ErrTaskNotFound = errors.New("task not found")

Functions

func NewRedisLocker

func NewRedisLocker(config LockerConfig, client redis.UniversalClient) gocron.Locker

func NewScheduler

func NewScheduler(config Config, jobSyncer Syncer, logger *zap.Logger, opts ...gocron.SchedulerOption) (gocron.Scheduler, error)

func NewSchedulerWithRedisLocker

func NewSchedulerWithRedisLocker(config Config, jobSyncer Syncer, client redis.UniversalClient, logger *zap.Logger, opts ...gocron.SchedulerOption) (gocron.Scheduler, error)

Types

type Config

type Config struct {
	// Limit sets the limit to be used by the Scheduler for limiting
	// the number of jobs that may be running at a given time.
	Limit uint `cfg:"limit,omitempty" json:"limit,omitempty" yaml:"limit,omitempty" bson:"limit,omitempty"`

	// Sync all the jobs
	Sync time.Duration `cfg:"sync,omitempty" json:"sync,omitempty" yaml:"sync,omitempty" bson:"sync,omitempty"`

	Locker      *LockerConfig `cfg:"locker,omitempty" json:"locker,omitempty" yaml:"locker,omitempty" bson:"locker,omitempty"`
	StopTimeout time.Duration `cfg:"stop_timeout,omitempty" json:"stop_timeout,omitempty" yaml:"stop_timeout,omitempty" bson:"stop_timeout,omitempty"`
}

func (*Config) InitDefaults

func (c *Config) InitDefaults()

type Job

type Job struct {
	// Name of the job aka ID, should be unique
	Name string `json:"name,omitempty"`
	// Type of the job, it's a string representation of JobType
	Type    JobType   `json:"type,omitempty"`
	Crontab string    `json:"crontab,omitempty"`
	Tags    []string  `json:"tags,omitempty"`
	Payload []byte    `json:"payload,omitempty"`
	Updated time.Time `json:"updated,omitempty"`
}

type JobSyncer

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

func NewSyncer

func NewSyncer(provider Provider, resolver Resolver, logger *zap.Logger) *JobSyncer

func (JobSyncer) Sync

func (syncer JobSyncer) Sync(ctx context.Context, s gocron.Scheduler)

type JobType

type JobType string

func (JobType) String

func (t JobType) String() string

type LockerConfig

type LockerConfig struct {
	// Tries can be used to set the number of times lock acquire is attempted.
	Tries int `cfg:"tries" json:"tries,omitempty" yaml:"tries,omitempty" bson:"tries,omitempty"`

	// DriftFactor can be used to set the clock drift factor.
	DriftFactor float64 `cfg:"drift_factor,omitempty" json:"drift_factor,omitempty" yaml:"drift_factor,omitempty" bson:"drift_factor,omitempty"`

	// TimeoutFactor can be used to set the timeout factor.
	TimeoutFactor float64 `` /* 126-byte string literal not displayed */

	// Expiry can be used to set the expiry of a mutex to the given value.
	Expiry time.Duration `cfg:"expiry,omitempty" json:"expiry,omitempty" yaml:"expiry,omitempty" bson:"expiry,omitempty"`

	// RetryDelay can be used to set the amount of time to wait between retries.
	RetryDelay time.Duration `cfg:"retry_delay,omitempty" json:"retry_delay,omitempty" yaml:"retry_delay,omitempty" bson:"retry_delay,omitempty"`

	// Value can be used to assign the random value without having to call lock.
	// This allows the ownership of a lock to be "transferred" and allows the lock to be unlocked from elsewhere.
	Value string `cfg:"value,omitempty" json:"value,omitempty" yaml:"value,omitempty" bson:"value,omitempty"`

	// FailFast can be used to quickly acquire and release the lock.
	// When some Redis servers are blocking, we do not need to wait for responses from all the Redis servers response.
	// As long as the quorum is met, we can assume the lock is acquired. The effect of this parameter is to achieve low
	// latency, avoid Redis blocking causing Lock/Unlock to not return for a long time.
	FailFast bool `cfg:"fail_fast,omitempty" json:"fail_fast,omitempty" yaml:"fail_fast,omitempty" bson:"fail_fast,omitempty"`

	// ShufflePools can be used to shuffle Redis pools to reduce centralized access in concurrent scenarios.
	ShufflePools bool `cfg:"shuffle_pools,omitempty" json:"shuffle_pools,omitempty" yaml:"shuffle_pools,omitempty" bson:"shuffle_pools,omitempty"`
}

func (*LockerConfig) InitDefaults

func (c *LockerConfig) InitDefaults()

type Logger

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

func (Logger) Debug

func (l Logger) Debug(msg string, args ...any)

func (Logger) Error

func (l Logger) Error(msg string, args ...any)

func (Logger) Info

func (l Logger) Info(msg string, args ...any)

func (Logger) Warn

func (l Logger) Warn(msg string, args ...any)

type MiddlewareFunc

type MiddlewareFunc func(next TaskFunc) TaskFunc

func MetricsMiddleware

func MetricsMiddleware() (MiddlewareFunc, error)

func TracingMiddleware

func TracingMiddleware() MiddlewareFunc

type Provider

type Provider interface {
	GetJobs(ctx context.Context, offset, size int) ([]Job, error)
}

type Resolver

type Resolver interface {
	Resolve(ctx context.Context, job Job) (Task, error)
}

type Syncer

type Syncer interface {
	Sync(ctx context.Context, s gocron.Scheduler)
}

type Task

type Task = gocron.Task

type TaskFunc

type TaskFunc func(ctx context.Context, job Job) error

type TaskResolver

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

func NewTaskResolver

func NewTaskResolver(tasks map[JobType]TaskFunc, middleware ...MiddlewareFunc) *TaskResolver

func (*TaskResolver) Resolve

func (r *TaskResolver) Resolve(ctx context.Context, job Job) (Task, error)