package cron import ( "log" "runtime" "sort" "time" ) // Cron keeps track of any number of entries, invoking the associated func as // specified by the schedule. It may be started, stopped, and the entries may // be inspected while running. type Cron struct { entries []*Entry stop chan struct{} add chan *Entry snapshot chan []*Entry running bool ErrorLog *log.Logger location *time.Location } // Job is an interface for submitted cron jobs. type Job interface { Run() } // The Schedule describes a job's duty cycle. type Schedule interface { // Return the next activation time, later than the given time. // Next is invoked initially, and then each time the job is run. Next(time.Time) time.Time } // Entry consists of a schedule and the func to execute on that schedule. type Entry struct { Description string Spec string // The schedule on which this job should be run. Schedule Schedule // The next time the job will run. This is the zero time if Cron has not been // started or this entry's schedule is unsatisfiable Next time.Time // The last time this job was run. This is the zero time if the job has never // been run. Prev time.Time // The Job to run. Job Job ExecTimes int // Execute times count. } // byTime is a wrapper for sorting the entry array by time // (with zero time at the end). type byTime []*Entry func (s byTime) Len() int { return len(s) } func (s byTime) Swap(i, j int) { s[i], s[j] = s[j], s[i] } func (s byTime) Less(i, j int) bool { // Two zero times should return false. // Otherwise, zero is "greater" than any other time. // (To sort it at the end of the list.) if s[i].Next.IsZero() { return false } if s[j].Next.IsZero() { return true } return s[i].Next.Before(s[j].Next) } // New returns a new Cron job runner, in the Local time zone. func New() *Cron { return NewWithLocation(time.Now().Location()) } // NewWithLocation returns a new Cron job runner. func NewWithLocation(location *time.Location) *Cron { return &Cron{ entries: nil, add: make(chan *Entry), stop: make(chan struct{}), snapshot: make(chan []*Entry), running: false, ErrorLog: nil, location: location, } } // A wrapper that turns a func() into a cron.Job type FuncJob func() func (f FuncJob) Run() { f() } // AddFunc adds a func to the Cron to be run on the given schedule. func (c *Cron) AddFunc(desc, spec string, cmd func()) (*Entry, error) { return c.AddJob(desc, spec, FuncJob(cmd)) } // AddJob adds a Job to the Cron to be run on the given schedule. func (c *Cron) AddJob(desc, spec string, cmd Job) (*Entry, error) { schedule, err := Parse(spec) if err != nil { return nil, err } return c.Schedule(desc, spec, schedule, cmd), nil } // Schedule adds a Job to the Cron to be run on the given schedule. func (c *Cron) Schedule(desc, spec string, schedule Schedule, cmd Job) *Entry { entry := &Entry{ Description: desc, Spec: spec, Schedule: schedule, Job: cmd, } if c.running { c.add <- entry } else { c.entries = append(c.entries, entry) } return entry } // Entries returns a snapshot of the cron entries. func (c *Cron) Entries() []*Entry { if c.running { c.snapshot <- nil x := <-c.snapshot return x } return c.entrySnapshot() } // Location gets the time zone location func (c *Cron) Location() *time.Location { return c.location } // Start the cron scheduler in its own go-routine, or no-op if already started. func (c *Cron) Start() { if c.running { return } c.running = true go c.run() } // Run the cron scheduler, or no-op if already running. func (c *Cron) Run() { if c.running { return } c.running = true c.run() } func (c *Cron) runWithRecovery(j Job) { defer func() { if r := recover(); r != nil { const size = 64 << 10 buf := make([]byte, size) buf = buf[:runtime.Stack(buf, false)] c.logf("cron: panic running job: %v\n%s", r, buf) } }() j.Run() } // Run the scheduler. this is private just due to the need to synchronize // access to the 'running' state variable. func (c *Cron) run() { // Figure out the next activation times for each entry. now := c.now() for _, entry := range c.entries { entry.Next = entry.Schedule.Next(now) } for { // Determine the next entry to run. sort.Sort(byTime(c.entries)) var timer *time.Timer if len(c.entries) == 0 || c.entries[0].Next.IsZero() { // If there are no entries yet, just sleep - it still handles new entries // and stop requests. timer = time.NewTimer(100000 * time.Hour) } else { timer = time.NewTimer(c.entries[0].Next.Sub(now)) } for { select { case now = <-timer.C: now = now.In(c.location) // Run every entry whose next time was less than now for _, e := range c.entries { if e.Next.After(now) || e.Next.IsZero() { break } go c.runWithRecovery(e.Job) e.ExecTimes++ e.Prev = e.Next e.Next = e.Schedule.Next(now) } case newEntry := <-c.add: timer.Stop() now = c.now() newEntry.Next = newEntry.Schedule.Next(now) c.entries = append(c.entries, newEntry) case <-c.snapshot: c.snapshot <- c.entrySnapshot() continue case <-c.stop: timer.Stop() return } break } } } // Logs an error to stderr or to the configured error log func (c *Cron) logf(format string, args ...interface{}) { if c.ErrorLog != nil { c.ErrorLog.Printf(format, args...) } else { log.Printf(format, args...) } } // Stop stops the cron scheduler if it is running; otherwise it does nothing. func (c *Cron) Stop() { if !c.running { return } c.stop <- struct{}{} c.running = false } // entrySnapshot returns a copy of the current cron entry list. func (c *Cron) entrySnapshot() []*Entry { entries := []*Entry{} for _, e := range c.entries { entries = append(entries, &Entry{ Description: e.Description, Spec: e.Spec, Schedule: e.Schedule, Next: e.Next, Prev: e.Prev, Job: e.Job, ExecTimes: e.ExecTimes, }) } return entries } // now returns current time in c location func (c *Cron) now() time.Time { return time.Now().In(c.location) }