diff options
author | Unknwon <u@gogs.io> | 2018-05-27 09:07:15 +0800 |
---|---|---|
committer | Unknwon <u@gogs.io> | 2018-05-27 09:07:15 +0800 |
commit | e33d9e77f43e6829ea967e47964d13f5a8aec5cc (patch) | |
tree | ace1a09414a66fd7e293b3837865633168ba4f6e /vendor/github.com/gogs/cron/cron.go | |
parent | aff42082441715559bb2e62e11550af1a946a8c9 (diff) |
vendor: rename "gogits" to "gogs"
Diffstat (limited to 'vendor/github.com/gogs/cron/cron.go')
-rw-r--r-- | vendor/github.com/gogs/cron/cron.go | 243 |
1 files changed, 243 insertions, 0 deletions
diff --git a/vendor/github.com/gogs/cron/cron.go b/vendor/github.com/gogs/cron/cron.go new file mode 100644 index 00000000..ed6ad21c --- /dev/null +++ b/vendor/github.com/gogs/cron/cron.go @@ -0,0 +1,243 @@ +// This library implements a cron spec parser and runner. See the README for +// more details. +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 +} + +// 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. +func New() *Cron { + return &Cron{ + entries: nil, + add: make(chan *Entry), + stop: make(chan struct{}), + snapshot: make(chan []*Entry), + running: false, + ErrorLog: nil, + } +} + +// 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() +} + +// 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() +} + +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 := time.Now().Local() + for _, entry := range c.entries { + entry.Next = entry.Schedule.Next(now) + } + + for { + // Determine the next entry to run. + sort.Sort(byTime(c.entries)) + + var effective time.Time + 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. + effective = now.AddDate(10, 0, 0) + } else { + effective = c.entries[0].Next + } + + timer := time.NewTimer(effective.Sub(now)) + select { + case now = <-timer.C: + // Run every entry whose next time was this effective time. + for _, e := range c.entries { + if e.Next != effective { + break + } + go c.runWithRecovery(e.Job) + e.ExecTimes++ + e.Prev = e.Next + e.Next = e.Schedule.Next(now) + } + continue + + case newEntry := <-c.add: + c.entries = append(c.entries, newEntry) + newEntry.Next = newEntry.Schedule.Next(time.Now().Local()) + + case <-c.snapshot: + c.snapshot <- c.entrySnapshot() + + case <-c.stop: + timer.Stop() + return + } + + // 'now' should be updated after newEntry and snapshot cases. + now = time.Now().Local() + timer.Stop() + } +} + +// 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 +} |