feat(cron.file): FileJobStore for saving jobs to a backing CBOR file

cron/cron.go

@@ -254,7 +254,7 @@ func (c *Cron) Unschedule(name string) error {
 // MemJobStore is an in-memory implementation of the JobStore interface
 // Jobs stored in MemJobStore can be persisted for the duration of a process
 // at the longest.
-// MemJobStore is save for concurrent use
+// MemJobStore is safe for concurrent use
 type MemJobStore struct {
  lock sync.Mutex
  jobs jobs

cron/cron_test.go

@@ -109,7 +109,8 @@ func CompareJobs(a, b *Job) error {
  if a.Periodicity != b.Periodicity {
  return fmt.Errorf("Periodicity mismatch. %s != %s", a.Name, b.Name)
  }
- if !a.LastRun.Equal(b.LastRun) {
+ // use unix comparisons to ignore millisecond & nanosecond precision errors
+ if a.LastRun.Unix() != b.LastRun.Unix() {
  return fmt.Errorf("LastRun mismatch. %s != %s", a.LastRun, b.LastRun)
  }
  if a.Type != b.Type {
@@ -156,7 +157,7 @@ func RunJobStoreTests(t *testing.T, newStore func() JobStore) {
  t.Fatal(err)
  }
  if len(jobs) != 1 {
- t.Errorf("expected default get to return inserted job")
+ t.Fatal("expected default get to return inserted job")
  }
  if err := CompareJobs(jobOne, jobs[0]); err != nil {
  t.Errorf("stored job mistmatch: %s", err)

cron/file.go

@@ -0,0 +1,161 @@
+package cron
+
+import (
+ "bytes"
+ "fmt"
+ "io/ioutil"
+ "os"
+ "sort"
+ "sync"
+
+ "github.com/ugorji/go/codec"
+)
+
+// NewFileJobStore creates a job store that persists to a CBOR file
+// specified at path
+func NewFileJobStore(path string) JobStore {
+ return &FileJobStore{
+ path: path,
+ }
+}
+
+// FileJobStore is a jobstore implementation that saves to a CBOR file
+// Jobs stored in FileJobStore can be persisted for the duration of a process
+// at the longest.
+// FileJobStore is safe for concurrent use
+type FileJobStore struct {
+ lock sync.Mutex
+ path string
+}
+
+// Jobs lists jobs currently in the store
+func (s *FileJobStore) Jobs(offset, limit int) ([]*Job, error) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+
+ js, err := s.loadJobs()
+ if err != nil {
+ return nil, err
+ }
+
+ if limit <= 0 {
+ limit = len(js)
+ }
+
+ ss := make([]*Job, limit)
+ added := 0
+ for i, job := range js {
+ if i < offset {
+ continue
+ } else if added == limit {
+ break
+ }
+
+ ss[added] = job
+ added++
+ }
+ return ss[:added], nil
+}
+
+func (s *FileJobStore) handle() codec.Handle {
+ return &codec.CborHandle{
+ // Need to use RFC3339 timestamps to preserve as much precision as possible
+ TimeRFC3339: true,
+ }
+}
+
+func (s *FileJobStore) loadJobs() (js jobs, err error) {
+ f, err := os.Open(s.path)
+ if os.IsNotExist(err) {
+ return jobs{}, nil
+ }
+
+ js = jobs{}
+ err = codec.NewDecoder(f, s.handle()).Decode(&js)
+ return
+}
+
+func (s *FileJobStore) saveJobs(jobs []*Job) error {
+ buf := &bytes.Buffer{}
+ if err := codec.NewEncoder(buf, s.handle()).Encode(jobs); err != nil {
+ return err
+ }
+ return ioutil.WriteFile(s.path, buf.Bytes(), os.ModePerm)
+}
+
+// Job gets job details from the store by name
+func (s *FileJobStore) Job(name string) (*Job, error) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+
+ js, err := s.loadJobs()
+ if err != nil {
+ return nil, err
+ }
+
+ for _, job := range js {
+ if job.Name == name {
+ return job, nil
+ }
+ }
+ return nil, fmt.Errorf("not found")
+}
+
+// PutJob places a job in the store. If the job name matches the name of a job
+// that already exists, it will be overwritten with the new job
+func (s *FileJobStore) PutJob(job *Job) error {
+ if err := ValidateJob(job); err != nil {
+ return err
+ }
+
+ s.lock.Lock()
+ defer s.lock.Unlock()
+
+ js, err := s.loadJobs()
+ if err != nil {
+ return err
+ }
+
+ for i, j := range js {
+ if job.Name == j.Name {
+ js[i] = job
+
+ sort.Sort(js)
+ return s.saveJobs(js)
+ }
+ }
+
+ js = append(js, job)
+ sort.Sort(js)
+ return s.saveJobs(js)
+}
+
+// DeleteJob removes a job from the store by name. deleting a non-existent job
+// won't return an error
+func (s *FileJobStore) DeleteJob(name string) error {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+
+ js, err := s.loadJobs()
+ if err != nil {
+ return err
+ }
+
+ for i, j := range js {
+ if j.Name == name {
+ if i+1 == len(js) {
+ js = js[:i]
+ break
+ }
+
+ js = append(js[:i], js[i+1:]...)
+ break
+ }
+ }
+ return s.saveJobs(js)
+}
+
+// Destroy removes the path entirely
+func (s *FileJobStore) Destroy() error {
+ return os.Remove(s.path)
+}

cron/file_test.go

@@ -0,0 +1,20 @@
+package cron
+
+import (
+ "io/ioutil"
+ "os"
+ "path/filepath"
+ "testing"
+)
+
+func TestFsJobStore(t *testing.T) {
+ tmp, err := ioutil.TempDir(os.TempDir(), "TestFsJobStore")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(tmp)
+ newStore := func() JobStore {
+ return NewFileJobStore(filepath.Join(tmp, "jobs.cbor"))
+ }
+ RunJobStoreTests(t, newStore)
+}