gopkg/gocron
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

export function comments

Browse Source
This commit is contained in:
John Roesler
2023-11-22 06:43:50 -06:00
parent 77dba0ad18
commit 63225ac1a9
7 changed files with 75 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.golangci.yaml
+3 -6
View File
@@ -7,12 +7,9 @@ run:
issues:
max-same-issues: 100
exclude-rules:
- path: _test\.go
linters:
- bodyclose
- errcheck
- gosec
include:
- EXC0012
- EXC0014
linters:
enable:
executor.go
+8 -8
View File
@@ -48,17 +48,17 @@ func (e *executor) start() {
e.ctx, e.cancel = context.WithCancel(context.Background())
// the standardJobsWg tracks
standardJobsWg := waitGroupWithMutex{
standardJobsWg := &waitGroupWithMutex{
wg: sync.WaitGroup{},
mu: sync.Mutex{},
}
singletonJobsWg := waitGroupWithMutex{
singletonJobsWg := &waitGroupWithMutex{
wg: sync.WaitGroup{},
mu: sync.Mutex{},
}
limitModeJobsWg := waitGroupWithMutex{
limitModeJobsWg := &waitGroupWithMutex{
wg: sync.WaitGroup{},
mu: sync.Mutex{},
}
@@ -78,7 +78,7 @@ func (e *executor) start() {
case id := <-e.jobsIDsIn:
select {
case <-e.stopCh:
e.stop(&standardJobsWg, &singletonJobsWg, &limitModeJobsWg)
e.stop(standardJobsWg, singletonJobsWg, limitModeJobsWg)
return
default:
}
@@ -92,7 +92,7 @@ func (e *executor) start() {
e.limitMode.started = true
for i := e.limitMode.limit; i > 0; i-- {
limitModeJobsWg.Add(1)
go e.limitModeRunner("limitMode-"+strconv.Itoa(int(i)), e.limitMode.in, &limitModeJobsWg, e.limitMode.mode, e.limitMode.rescheduleLimiter)
go e.limitModeRunner("limitMode-"+strconv.Itoa(int(i)), e.limitMode.in, limitModeJobsWg, e.limitMode.mode, e.limitMode.rescheduleLimiter)
}
}
@@ -155,7 +155,7 @@ func (e *executor) start() {
}
e.singletonRunners[id] = runner
singletonJobsWg.Add(1)
go e.limitModeRunner("singleton-"+id.String(), runner.in, &singletonJobsWg, j.singletonLimitMode, runner.rescheduleLimiter)
go e.limitModeRunner("singleton-"+id.String(), runner.in, singletonJobsWg, j.singletonLimitMode, runner.rescheduleLimiter)
}
if j.singletonLimitMode == LimitModeReschedule {
@@ -180,7 +180,7 @@ func (e *executor) start() {
} else {
select {
case <-e.stopCh:
e.stop(&standardJobsWg, &singletonJobsWg, &limitModeJobsWg)
e.stop(standardJobsWg, singletonJobsWg, limitModeJobsWg)
return
default:
}
@@ -198,7 +198,7 @@ func (e *executor) start() {
}
}()
case <-e.stopCh:
e.stop(&standardJobsWg, &singletonJobsWg, &limitModeJobsWg)
e.stop(standardJobsWg, singletonJobsWg, limitModeJobsWg)
return
}
}
job.go
+41 -2
View File
@@ -196,6 +196,12 @@ func DurationRandomJob(minDuration, maxDuration time.Duration) JobDefinition {
}
}
// DailyJob runs the job on the interval of days, and at the set times.
// By default, the job will start the next available day, considering the last run to be now,
// and the time and day based on the interval and times you input. This means, if you
// select an interval greater than 1, your job by default will run X (interval) days from now
// if there are no atTimes left in the current day. You can use WithStartAt to tell the
// scheduler to start the job sooner.
func DailyJob(interval uint, atTimes AtTimes) JobDefinition {
return dailyJobDefinition{
interval: interval,
@@ -268,14 +274,24 @@ func (w weeklyJobDefinition) setup(j *internalJob, location *time.Location) erro
return nil
}
// Weekdays defines a function that returns a list of week days.
type Weekdays func() []time.Weekday
// NewWeekdays provide the days of the week the job should run.
func NewWeekdays(weekday time.Weekday, weekdays ...time.Weekday) Weekdays {
return func() []time.Weekday {
return append(weekdays, weekday)
}
}
// WeeklyJob runs the job on the interval of weeks, on the specific days of the week
// specified, and at the set times.
//
// By default, the job will start the next available day, considering the last run to be now,
// and the time and day based on the interval, days and times you input. This means, if you
// select an interval greater than 1, your job by default will run X (interval) weeks from now
// if there are no daysOfTheWeek left in the current week. You can use WithStartAt to tell the
// scheduler to start the job sooner.
func WeeklyJob(interval uint, daysOfTheWeek Weekdays, atTimes AtTimes) JobDefinition {
return weeklyJobDefinition{
interval: interval,
@@ -395,7 +411,7 @@ func NewAtTimes(atTime AtTime, atTimes ...AtTime) AtTimes {
// By default, the job will start the next available day, considering the last run to be now,
// and the time and month based on the interval, days and times you input.
// This means, if you select an interval greater than 1, your job by default will run
// X (interval) months from now.
// X (interval) months from now if there are no daysOfTheMonth left in the current month.
// You can use WithStartAt to tell the scheduler to start the job sooner.
//
// Carefully consider your configuration!
@@ -415,8 +431,11 @@ func MonthlyJob(interval uint, daysOfTheMonth DaysOfTheMonth, atTimes AtTimes) J
// -----------------------------------------------
// -----------------------------------------------
// JobOption defines the constructor for job options.
type JobOption func(*internalJob) error
// WithEventListeners sets the event listeners that should be
// run for the job.
func WithEventListeners(eventListeners ...EventListener) JobOption {
return func(j *internalJob) error {
for _, eventListener := range eventListeners {
@@ -453,6 +472,13 @@ func WithName(name string) JobOption {
}
}
// WithSingletonMode keeps the job from running again if it is already running.
// This is useful for jobs that should not overlap, and that occasionally
// (but not consistently) run longer than the interval between job runs.
//
// Note - this is mutually exclusive with WithLimitConcurrentJobs. If both
// are set, WithLimitConcurrentJobs will take precedence.
// WithSingletonMode effectively sets a per-job limit of 1 concurrent job.
func WithSingletonMode(mode LimitMode) JobOption {
return func(j *internalJob) error {
j.singletonMode = true
@@ -461,7 +487,8 @@ func WithSingletonMode(mode LimitMode) JobOption {
}
}
// WithStartAt sets the option for starting the job
// WithStartAt sets the option for starting the job at
// a specific datetime.
func WithStartAt(option StartAtOption) JobOption {
return func(j *internalJob) error {
return option(j)
@@ -482,6 +509,7 @@ func WithStartImmediately() StartAtOption {
}
// WithStartDateTime sets the first date & time at which the job should run.
// This datetime must be in the future.
func WithStartDateTime(start time.Time) StartAtOption {
return func(j *internalJob) error {
if start.IsZero() || start.Before(time.Now()) {
@@ -492,6 +520,9 @@ func WithStartDateTime(start time.Time) StartAtOption {
}
}
// WithTags sets the tags for the job. Tags provide
// a way to identify jobs by a set of tags and remove
// multiple jobs by tag.
func WithTags(tags ...string) JobOption {
return func(j *internalJob) error {
j.tags = tags
@@ -505,8 +536,12 @@ func WithTags(tags ...string) JobOption {
// -----------------------------------------------
// -----------------------------------------------
// EventListener defines the constructor for event
// listeners that can be used to listen for job events.
type EventListener func(*internalJob) error
// AfterJobRuns is used to listen for when a job has run regardless
// of any returned error value, and run the provided function.
func AfterJobRuns(eventListenerFunc func(jobID uuid.UUID, jobName string)) EventListener {
return func(j *internalJob) error {
if eventListenerFunc == nil {
@@ -517,6 +552,8 @@ func AfterJobRuns(eventListenerFunc func(jobID uuid.UUID, jobName string)) Event
}
}
// AfterJobRunsWithError is used to listen for when a job has run and
// returned an error, and then run the provided function.
func AfterJobRunsWithError(eventListenerFunc func(jobID uuid.UUID, jobName string, err error)) EventListener {
return func(j *internalJob) error {
if eventListenerFunc == nil {
@@ -527,6 +564,8 @@ func AfterJobRunsWithError(eventListenerFunc func(jobID uuid.UUID, jobName strin
}
}
// BeforeJobRuns is used to listen for when a job is about to run and
// then run the provided function.
func BeforeJobRuns(eventListenerFunc func(jobID uuid.UUID, jobName string)) EventListener {
return func(j *internalJob) error {
if eventListenerFunc == nil {
logger.go
+1
View File
@@ -36,6 +36,7 @@ var _ Logger = (*logger)(nil)
// when using the basic NewLogger.
type LogLevel int
// The different log levels that can be used.
const (
LogLevelError LogLevel = iota
LogLevelWarn
scheduler.go
+11
View File
@@ -14,6 +14,7 @@ import (
var _ Scheduler = (*scheduler)(nil)
// Scheduler defines the interface for the Scheduler.
type Scheduler interface {
Jobs() []Job
NewJob(JobDefinition, Task, ...JobOption) (Job, error)
@@ -62,6 +63,11 @@ type allJobsOutRequest struct {
outChan chan []Job
}
// NewScheduler creates a new Scheduler instance.
// The Scheduler is not started until Start() is called.
//
// NewJob will add jobs to the Scheduler, but they will not
// be scheduled until Start() is called.
func NewScheduler(options ...SchedulerOption) (Scheduler, error) {
schCtx, cancel := context.WithCancel(context.Background())
@@ -599,6 +605,10 @@ const (
// WithLimitConcurrentJobs sets the limit and mode to be used by the
// Scheduler for limiting the number of jobs that may be running at
// a given time.
//
// Note - this is mutually exclusive with WithSingletonMode. If both
// are set, WithLimitConcurrentJobs will take precedence.
// WithSingletonMode effectively sets a per-job limit of 1 concurrent job.
func WithLimitConcurrentJobs(limit uint, mode LimitMode) SchedulerOption {
return func(s *scheduler) error {
if limit == 0 {
@@ -629,6 +639,7 @@ func WithLocation(location *time.Location) SchedulerOption {
}
}
// WithLogger sets the logger to be used by the Scheduler.
func WithLogger(logger Logger) SchedulerOption {
return func(s *scheduler) error {
if logger == nil {
scheduler_test.go
+9
View File
@@ -1192,6 +1192,15 @@ func TestScheduler_WithEventListeners(t *testing.T) {
true,
testErr,
},
{
"AfterJobRunsWithError - multiple return values, including error",
NewTask(func() (bool, error) { return false, testErr }),
AfterJobRunsWithError(func(_ uuid.UUID, _ string, err error) {
listenerRunCh <- err
}),
true,
testErr,
},
{
"AfterJobRunsWithError - no error",
NewTask(func() error { return nil }),
util.go
+2 -2
View File
@@ -26,8 +26,8 @@ func callJobFuncWithParams(jobFunc any, params ...any) error {
for k, param := range params {
in[k] = reflect.ValueOf(param)
}
vals := f.Call(in)
for _, val := range vals {
returnValues := f.Call(in)
for _, val := range returnValues {
i := val.Interface()
if err, ok := i.(error); ok {
return err