jobs

README

jobs

Job management for MongoDB (via mgo).

• Simple API
• Jobs guaranteed to run only once (no matter how many runners)
• Retries
• Run jobs ASAP, or at a future date
• MongoDB as its backend

Usage

// create a job and set some data
job := jobs.New("notifications")
job.Data["message"] = "Hello world"

// put the job
jobs.Put(db.C("jobs"), job)

Meanwhile, in a process far, far away:

r := jobs.NewRunner("runner-1", db.C("jobs"), "notifications", func(j *jobs.J) error {
	log.Println("TODO: process this message -", j.Data["message"])
	return nil
})
r.Start()

Documentation

Overview

Package jobs provides job management for MongoDB via mgo.

Index

• func Put(c *mgo.Collection, jobs ...*J) error
• type J
  • func New(kind string) *J
• type JobFunc
• type Runner
  • func NewRunner(name string, c *mgo.Collection, kind string, fn JobFunc) *Runner
  • func (r *Runner) Err() error
  • func (r *Runner) Kind() string
  • func (r *Runner) Name() string
  • func (r *Runner) Start() error
  • func (r *Runner) Stop()
  • func (r *Runner) StopChan() <-chan struct{}
• type Status
  • func (s Status) MarshalText() (text []byte, err error)
  • func (s Status) String() string
  • func (s *Status) UnmarshalText(text []byte) error
• type Try

Functions

func Put

func Put(c *mgo.Collection, jobs ...*J) error

Put adds the jobs to the specified collection.

Types

type J

type J struct {
	// id is the unique job id.
	ID bson.ObjectId `bson:"_id" json:"id"`
	// Status is the current status of the job.
	Status Status `json:"status"`
	// Tries holds history of attempts.
	Tries []*Try `json:"tries"`
	// Created is when the job was created.
	Created time.Time `json:"created"`
	// RunAt is the time this job should run at (or after).
	RunAt time.Time `json:"runat" bson:"runat"`
	// Data is the user data for this job.
	Data map[string]interface{} `json:"data" bson:"data"`
	// Retries is the number of remaining attempts that will
	// be made to run this job.
	Retries int
	// RetryInterval is the time to wait after a failure before
	// trying to run the job again.
	RetryInterval time.Duration
	// Kind is the kind for this job. Only runners with the same kind
	// will be asked to process this job.
	Kind string
}

J is a job.

func New

func New(kind string) *J

New creates a new job with the specified kind.

type JobFunc

type JobFunc func(job *J) error

JobFunc is the function that gets called for each job.

type Runner

type Runner struct {
	Interval time.Duration
	// contains filtered or unexported fields
}

Runner runs jobs.

func NewRunner

func NewRunner(name string, c *mgo.Collection, kind string, fn JobFunc) *Runner

NewRunner makes a new Runner capable of running jobs. The name should be unique across a system. The mgo.Collection is where the job records live in MongoDB. The kind string should match J.Kind values for jobs that this runner should execute.

func (*Runner) Err

func (r *Runner) Err() error

Err is the last error that occurred.

func (*Runner) Kind

func (r *Runner) Kind() string

Kind is the name of the types of jobs this runner will process.

func (*Runner) Name

func (r *Runner) Name() string

Name is the name of the runner.

func (*Runner) Start

func (r *Runner) Start() error

Start starts the process.

func (*Runner) Stop

func (r *Runner) Stop()

Stop stops the runner. Callers should then block on StopChan() to be notified of when the runner has stopped.

func (*Runner) StopChan

func (r *Runner) StopChan() <-chan struct{}

StopChan is a channel that gets closed when the runner has stopped. Callers should block on this after calling Stop to ensure the runner has properly stopped.

<-runner.StopChan()

type Status

type Status int8

Status is the status of a job.

const (
	// StatusInvalid represents invalid status values.
	StatusInvalid Status = iota
	// StatusNew means a job has just been created.
	StatusNew
	// StatusWorking means the job is being worked on.
	StatusWorking
	// StatusWaiting means the job is waiting to retry.
	StatusWaiting
	// StatusSuccess means the job was successful.
	StatusSuccess
	// StatusFailed means the job failed.
	StatusFailed
)

func (Status) MarshalText

func (s Status) MarshalText() (text []byte, err error)

MarshalText marshals the value into bytes.

func (Status) String

func (s Status) String() string

String gets the string for the status.

func (*Status) UnmarshalText

func (s *Status) UnmarshalText(text []byte) error

UnmarshalText unmarshals the bytes into the value.

type Try

type Try struct {
	// Runner is the name of the runner that tried to
	// run this job.
	Runner string `json:"runner" bson:"runner"`
	// When is a timestamp of when the attempt took place.
	When time.Time `json:"when" bson:"when"`
	// Err is the error that was returned by JobFunc.
	Err string `json:"err,omitempty" bson:"err,omitempty"`
}

Try contains details of an attempt to run a job.