README
¶
metre
Golang cron framework. Scheduling through ØMQ push-pull and state managed through redis
Requirements
Installation
go get github.com/johnhof/metre
Documentation
- API References
Example
// Package main demo's metre
package main
import (
"github.com/johnhof/metre"
)
func main () {
// Create
m, _ := metre.New("127.0.0.1:5555", "127.0.0.1:6379") // (zmq, redis)
// Add tasks
m.add(metre.Task{
ID: "F", // used to prevent collision across tasks
Interval: "* * * * * *", // Cron expression
Schedule: func(t metre.TaskConfig, s metre.Scheduler, c metre.cache, q metre.queue) {
t.UID = "TestID" // overwrite the generated UID with a static namespace
s.Schedule(t) // only schedules if "TestID" is not being processed ("F-TestId" not cached in a processing state)
s.SetExpire(t, 15000) // set expire in milliseconds
},
Process: func(t metre.TaskConfig, s metre.Scheduler, c metre.Cache, q metre.Queue) {
log.Info("Processing: " + t.class + "-" + t.uid)
})
m.add(metre.Task{
ID: "B", // used to prevent collision across tasks
Interval: "@every minute", // Cron Schedule
Schedule: func(s metre.Scheduler, c metre.Cache, q metre.Queue) {
t.UID = "TestID" // overwrite the generated UID with a static namespace
s.ForceSchedule(t) // schedule regardless of task state
},
Process: func(t metre.TaskInstance, s metre.Scheduler, c metre.Cache, q metre.Queue) {
log.Info("Processing: " + t.class + "-" + t.uid)
},
})
go m.RunSlave() // run slave in background
m.RunMaster() // run master
// this approach means we can run master and slave
// on separate servers from the same code base
// allowing us to scale them separately
}
Authors
Documentation
¶
Overview ¶
Package metre is used to schedule end execute cron jobs in a simplified fashion
Index ¶
Constants ¶
const CACHEPORT string = "6379" // Default port for cache
const LOCALHOST string = "127.0.0.1" // Default host for cache and queue
const QUEUEPORT string = "5555" // Default port for queue
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Cache ¶
type Metre ¶
type Metre struct {
Cron cron.Cron
Queue Queue
Cache Cache
Scheduler Scheduler
TaskMap map[string]Task
}
func (*Metre) StartMaster ¶
func (m *Metre) StartMaster()
func (*Metre) StartSlave ¶
func (m *Metre) StartSlave()
type Queue ¶
type Queue struct {
URI string
PushSocket *zmq.Socket // Socket to push messages to
PullSocket *zmq.Socket // SOcket to pull mesasges form
}
func (Queue) ConnectPull ¶
ConnectPull connects to the socket to listen for messages
type Scheduler ¶
func NewScheduler ¶
func (Scheduler) ForceSchedule ¶
func (s Scheduler) ForceSchedule(t TaskRecord) (string, error)
ForceSchedule schedules a task in the cache and queue regardless of tasks actively waiting to be processed
func (Scheduler) Schedule ¶
func (s Scheduler) Schedule(t TaskRecord) (string, error)
Schedule schedules a task in the cache and queue if no task is actively waiting to be processed
func (Scheduler) SetExpire ¶
func (s Scheduler) SetExpire(t TaskRecord, time int)
SetExpire set teh expiration for a task
type Task ¶
type Task struct {
ID string // Type Type of task (user as class prefix in cache)
Interval string // Schedule String in cron notation
Schedule func(t TaskRecord, s Scheduler, c Cache, q Queue)
Process func(t TaskRecord, s Scheduler, c Cache, q Queue)
}
type TaskRecord ¶
func NewTaskRecord ¶
func NewTaskRecord(id string) TaskRecord
NewTaskRecord takes seed data and returns a full TaskRecord instance
func NewTaskRecordWithID ¶
func NewTaskRecordWithID(id string, uid string) TaskRecord
NewTaskRecordWithID takes seed data and returns a full TaskRecord instance
func ParseTask ¶
func ParseTask(s string) (TaskRecord, error)
ParseTask returns a task instance from a serialized json string
func (*TaskRecord) CanReschedule ¶
func (t *TaskRecord) CanReschedule() bool
CanReschedule returns whether or not the task can be rescheduled
func (*TaskRecord) SetProcessing ¶
func (t *TaskRecord) SetProcessing()
SetProcessing set the state of the task to "PROCESSING"
func (*TaskRecord) SetScheduled ¶
func (t *TaskRecord) SetScheduled()
SetProcessing set the state of the task to "SCHEDULED"
func (*TaskRecord) SetUnscheduled ¶
func (t *TaskRecord) SetUnscheduled()
SetProcessing set the state of the task to "UNSCHEDULED"
func (TaskRecord) ToString ¶
func (t TaskRecord) ToString() (string, error)
ToString returns a serialized json string representation of the task
func (*TaskRecord) TriggerUpdate ¶
func (t *TaskRecord) TriggerUpdate()
TriggerUpdate sets the UpdatedAt property to teh current time
Source Files