horloge

README

Horloge

Rationale

Getting started

Launch Horloge using docker

$ docker pull shinuza/horloge
$ docker run -p 127.0.0.1:6432:6432/tcp shinuza/horloge

Using the CLI

$ go get github.com/marjordome/horloge
$ horloge --bind 0.0.0.0 --port 1234

CLI Usage

NAME:
   horloge - A new cli application

USAGE:
   main.exe [global options] command [command options] [arguments...]

VERSION:
   0.1.0

AUTHOR:
   Samori Gorse <samorigorse@gail.com>

COMMANDS:
     help, h  Shows a list of commands or help for one command

GLOBAL OPTIONS:
   --port value, -p value  Port to listen to (default: 6432)
   --bind value, -b value  Address to bind to (default: "127.0.0.1")
   --help, -h              show help
   --version, -v           print the version

Launch a Horloge task manager

package main

import (
	"fmt"
	"time"

	"github.com/majordome/horloge"
)

func main() {
	runner := horloge.NewRunner()
	pattern := horloge.NewPattern("daily").At(9, 30, 0)
	job := horloge.NewJob("wake up", *pattern)

	runner.AddJob(job)

	runner.AddHandler("wake up", func(name string, args []string, t time.Time) {
		fmt.Printf("[INFO] Running \"%s\" with args %+v at %s\n", name, args, t.String())
	})

	select {}
}

What it is

Horloge is a cron like task runner

What it isn't

Horloge is not an asyncronous task runner like Celery or Faktory.

Documentation

Index

• Constants
• func Bod(t time.Time) time.Time
• func Bom(t time.Time) time.Time
• func Boy(t time.Time) time.Time
• func DiffToMonth(t time.Time, m time.Month) int
• func DiffToWeekday(t time.Time, wd time.Weekday) int
• func HTTPHandlerDeleteJob(r *Runner) func(c *gin.Context)
• func HTTPHandlerHealthCheck() func(c *gin.Context)
• func HTTPHandlerJobDetail(r *Runner) func(c *gin.Context)
• func HTTPHandlerListJobs(r *Runner) func(c *gin.Context)
• func HTTPHandlerPing() func(c *gin.Context)
• func HTTPHandlerRegisterJob(r *Runner) func(c *gin.Context)
• func HTTPHandlerVersion() func(c *gin.Context)
• func JobArgs(a []interface{}) (string, []string, time.Time)
• type Callback
• type Event
• type JSONMessage
• type Job
  • func NewJob(name string, pattern Pattern, args []string) *Job
  • func (j *Job) Bind(args []string)
  • func (j *Job) Calendar() []time.Time
  • func (j *Job) Cancel()
  • func (j *Job) Repeat() []time.Time
• type JobScheduledMessage
• type MQTTClient
  • func NewMQTTClient(addr string) *MQTTClient
  • func (c *MQTTClient) Publish(channel string, payload string)
• type Pattern
  • func NewPattern(occurence string) *Pattern
  • func (p *Pattern) At(hour, minute, second int) *Pattern
  • func (p *Pattern) IsZero() bool
  • func (p *Pattern) On(repeaters ...interface{}) *Pattern
  • func (p *Pattern) Time() time.Time
• type PublishHandler
• type RedisClient
  • func NewRedisClient(addr string, passwd string, db int) *RedisClient
  • func (c *RedisClient) AddPublishHandler(handler PublishHandler)
  • func (c *RedisClient) Wait()
• type Runner
  • func NewRunner() *Runner
  • func (r *Runner) AddHandler(name string, f Callback)
  • func (r *Runner) AddJob(job Job) ([]time.Time, error)
  • func (r *Runner) AddJobs(jobs []Job)
  • func (r *Runner) Emit(name string, args ...interface{})
  • func (r *Runner) GetJob(name string) *Job
  • func (r *Runner) HasJob(job *Job) bool
  • func (r *Runner) RemoveJob(job *Job)
  • func (r *Runner) Schedule(j Job, n time.Time, index int)
  • func (r *Runner) Subscribe(name string, f Callback)
  • func (r *Runner) Sync(s Sync)
  • func (r *Runner) ToJSON() ([]*Job, error)
• type Server
  • func NewWebsocketServer() *Server
  • func (s *Server) Publish(message string)
  • func (s *Server) Run(addr string)
• type Sync
• type SyncRedis
  • func NewSyncRedis(runner *Runner, addr string, password string, db int) *SyncRedis
  • func (s *SyncRedis) Read() []Job
  • func (s *SyncRedis) Write(jobs []*Job) error

Constants

const (
	MalformedMessage      string = "Malformed or empty request body"
	InvalidJobRequestBody string = "\"pattern\" and \"name\" must be present"
	UnableToSerializeJobs string = "Unable to serialize jobs"
)

const HORLOGE_KEY = "horloge_jobs"

const (
	JobExistsError string = "Cannot add task \"%s\", another task with the same name exists"
)

const MQTT_CHANNEL = "horloge/job"

const PUBSUB_CHANNEL = "horloge"

const (
	Version string = "0.1.0"
)

Functions

func Bod

func Bod(t time.Time) time.Time

func Bom

func Bom(t time.Time) time.Time

func Boy

func Boy(t time.Time) time.Time

func DiffToMonth

func DiffToMonth(t time.Time, m time.Month) int

func DiffToWeekday

func DiffToWeekday(t time.Time, wd time.Weekday) int

func HTTPHandlerDeleteJob

func HTTPHandlerDeleteJob(r *Runner) func(c *gin.Context)

HTTPHandlerDeleteJob Delete a job

func HTTPHandlerHealthCheck

func HTTPHandlerHealthCheck() func(c *gin.Context)

HTTPHandlerHealthCheck Handles GET requests to /health_check.

Replies an empty string and status code 200. This is useful if you want to monitor the state of the application.

func HTTPHandlerJobDetail

func HTTPHandlerJobDetail(r *Runner) func(c *gin.Context)

HTTPHandlerJobDetail Show job detail

func HTTPHandlerListJobs

func HTTPHandlerListJobs(r *Runner) func(c *gin.Context)

HTTPHandlerListJobs List Jobs

func HTTPHandlerPing

func HTTPHandlerPing() func(c *gin.Context)

HTTPHandlerPing Handles GET requests to /ping.

Replies with pong.

func HTTPHandlerRegisterJob

func HTTPHandlerRegisterJob(r *Runner) func(c *gin.Context)

HTTPHandlerRegisterJob Handles POST requests to /jobs.

To add a job you must send a request to /jobs with a json body

func HTTPHandlerVersion

func HTTPHandlerVersion() func(c *gin.Context)

HTTPHandlerVersion Handles GET requests to /version.

Replies with project version

func JobArgs

func JobArgs(a []interface{}) (string, []string, time.Time)

Types

type Callback

type Callback func(...interface{})

type Event

type Event struct {
	Name string    `json:"name"`
	Args []string  `json:"args"`
	Time time.Time `json:"time"`
}

type JSONMessage

type JSONMessage struct {
	Message string `json:"message"`
	Details string `json:"details"`
}

type Job

type Job struct {
	Name        string
	Description string
	Args        []string
	Pattern     Pattern
	// contains filtered or unexported fields
}

func NewJob

func NewJob(name string, pattern Pattern, args []string) *Job

func (*Job) Bind

func (j *Job) Bind(args []string)

func (*Job) Calendar

func (j *Job) Calendar() []time.Time

func (*Job) Cancel

func (j *Job) Cancel()

func (*Job) Repeat

func (j *Job) Repeat() []time.Time

type JobScheduledMessage

type JobScheduledMessage struct {
	Nexts []time.Time `json:"nexts"`
	Name  string      `json:"name"`
}

type MQTTClient

type MQTTClient struct {
	// contains filtered or unexported fields
}

func NewMQTTClient

func NewMQTTClient(addr string) *MQTTClient

func (*MQTTClient) Publish

func (c *MQTTClient) Publish(channel string, payload string)

type Pattern

type Pattern struct {
	Second    int    `json:"second" form:"second" query:"second"`
	Minute    int    `json:"minute" form:"minute" query:"minute"`
	Hour      int    `json:"hour" form:"hour" query:"hour"`
	Day       int    `json:"day" form:"day" query:"day"`
	Month     int    `json:"month" form:"month" query:"month"`
	Year      int    `json:"year" form:"year" query:"year"`
	Occurence string `json:"occurence" form:"occurence" query:"occurence"`
	Days      []time.Weekday
	Months    []time.Month
	Now       time.Time
}

func NewPattern

func NewPattern(occurence string) *Pattern

func (*Pattern) At

func (p *Pattern) At(hour, minute, second int) *Pattern

func (*Pattern) IsZero

func (p *Pattern) IsZero() bool

func (*Pattern) On

func (p *Pattern) On(repeaters ...interface{}) *Pattern

func (*Pattern) Time

func (p *Pattern) Time() time.Time

type PublishHandler

type PublishHandler func(*redis.Message)

type RedisClient

type RedisClient struct {
	// contains filtered or unexported fields
}

func NewRedisClient

func NewRedisClient(addr string, passwd string, db int) *RedisClient

func (*RedisClient) AddPublishHandler

func (c *RedisClient) AddPublishHandler(handler PublishHandler)

func (*RedisClient) Wait

func (c *RedisClient) Wait()

type Runner

type Runner struct {
	// contains filtered or unexported fields
}

func NewRunner

func NewRunner() *Runner

func (*Runner) AddHandler

func (r *Runner) AddHandler(name string, f Callback)

func (*Runner) AddJob

func (r *Runner) AddJob(job Job) ([]time.Time, error)

func (*Runner) AddJobs

func (r *Runner) AddJobs(jobs []Job)

func (*Runner) Emit

func (r *Runner) Emit(name string, args ...interface{})

func (*Runner) GetJob

func (r *Runner) GetJob(name string) *Job

func (*Runner) HasJob

func (r *Runner) HasJob(job *Job) bool

func (*Runner) RemoveJob

func (r *Runner) RemoveJob(job *Job)

func (*Runner) Schedule

func (r *Runner) Schedule(j Job, n time.Time, index int)

func (*Runner) Subscribe

func (r *Runner) Subscribe(name string, f Callback)

func (*Runner) Sync

func (r *Runner) Sync(s Sync)

func (*Runner) ToJSON

func (r *Runner) ToJSON() ([]*Job, error)

type Server

type Server struct {
	// contains filtered or unexported fields
}

func NewWebsocketServer

func NewWebsocketServer() *Server

func (*Server) Publish

func (s *Server) Publish(message string)

func (*Server) Run

func (s *Server) Run(addr string)

type Sync

type Sync interface {
	Write([]*Job) error
	Read() []Job
}

type SyncRedis

type SyncRedis struct {
	Client *redis.Client

	Addr     string
	Password string
	DB       int
	// contains filtered or unexported fields
}

func NewSyncRedis

func NewSyncRedis(runner *Runner, addr string, password string, db int) *SyncRedis

func (*SyncRedis) Read

func (s *SyncRedis) Read() []Job

func (*SyncRedis) Write

func (s *SyncRedis) Write(jobs []*Job) error