red

README

go-red

Golang distributed job scheduling.

Installation

go-red may be installed using the go get command:

go get github.com/pghq/go-red

Usage

import "github.com/pghq/go-red"

To create a new queue:

queue := red.New("redis://user:pass@example.com?queue=messages")
if err := queue.Enqueue(context.TODO(), "key", "value"); err != nil{
    panic(err)
}

Powered By

• Redis - https://redis.io

Documentation

Index

• Constants
• Variables
• type Log
  • func (l Log) Debugf(format string, args ...interface{})
  • func (l Log) Infof(format string, args ...interface{})
• type Message
  • func (m *Message) Ack(ctx context.Context) error
  • func (m *Message) Decode(v interface{}) error
  • func (m *Message) Reject(ctx context.Context) error
• type Red
  • func New(redisURL string) *Red
  • func (r *Red) Dequeue(ctx context.Context) (*Message, error)
  • func (r *Red) Enqueue(ctx context.Context, key, value interface{}) error
  • func (r Red) Error() error
  • func (r Red) Lock(name string, opts ...redsync.Option) *redsync.Mutex
  • func (r Red) Once(key string)
  • func (r Red) RLock(name string, opts ...redsync.Option) *redsync.Mutex
  • func (r Red) Repeat(key, recurrence string) error
  • func (r Red) StartScheduling(handler func(task *Task), schedulers ...func())
  • func (r Red) StopScheduling()
  • func (r Red) Wait()
• type Scheduler
  • func NewScheduler(queue *Red) *Scheduler
  • func (s *Scheduler) Add(tasks ...*Task) *Scheduler
  • func (s *Scheduler) DequeueTimeout(timeout time.Duration) *Scheduler
  • func (s *Scheduler) EnqueueTimeout(timeout time.Duration) *Scheduler
  • func (s *Scheduler) Every(interval time.Duration) *Scheduler
  • func (s *Scheduler) Handle(fn func(task *Task))
  • func (s *Scheduler) NewTask(id string) *Task
  • func (s *Scheduler) Start()
  • func (s *Scheduler) Stop()
• type Task
  • func (t *Task) CanSchedule(now time.Time) bool
  • func (t *Task) IsComplete() bool
  • func (t *Task) MarkScheduled(at time.Time) *Task
  • func (t *Task) Occurrences() int
  • func (t *Task) SetRecurrence(rfc string) error
• type TaskSchedule
• type Worker
  • func NewQWorker(name string, jobs ...func()) *Worker
  • func NewWorker(name string, jobs ...func()) *Worker
  • func (w *Worker) AddJobs(jobs ...func())
  • func (w *Worker) Concurrent(instances int) *Worker
  • func (w *Worker) Every(interval time.Duration) *Worker
  • func (w *Worker) Start()
  • func (w *Worker) Stop()

Constants

const (
	// DefaultSchedulerInterval is the default period between checking tasks to schedule
	DefaultSchedulerInterval = time.Millisecond

	// DefaultEnqueueTimeout is the default time allowed for queue write ops
	DefaultEnqueueTimeout = 100 * time.Millisecond

	// DefaultDequeueTimeout is the default time allowed for queue read ops
	DefaultDequeueTimeout = 100 * time.Millisecond
)

const (
	// DefaultInstances is the default number of simultaneous workers
	DefaultInstances = 1

	// DefaultWorkerInterval is the default period between running batches of jobs
	DefaultWorkerInterval = time.Millisecond
)

const (
	// BatchSize the total number of errors and messages consumed before dropping occurs
	BatchSize = 1024
)

Variables

var (
	// ErrNoMessages is an error returned when a request is made but no messages are available
	ErrNoMessages = trail.NewErrorBadRequest("no messages currently available")
)

Types

type Log

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

Log instance with quiet support

func (Log) Debugf

func (l Log) Debugf(format string, args ...interface{})

Debugf formatted value

func (Log) Infof

func (l Log) Infof(format string, args ...interface{})

Infof formatted value

type Message

type Message struct {
	Id    string `json:"id"`
	Value []byte `json:"value"`
	// contains filtered or unexported fields
}

Message is a single instance of a message within the queue.

func (*Message) Ack

func (m *Message) Ack(ctx context.Context) error

Ack notifies upstream of successful message handling

func (*Message) Decode

func (m *Message) Decode(v interface{}) error

Decode fills the supplied interface with the underlying message value

func (*Message) Reject

func (m *Message) Reject(ctx context.Context) error

Reject notifies upstream of unsuccessful message handling

type Red

type Red struct {
	Name string
	URL  *url.URL
	// contains filtered or unexported fields
}

Red is an instance of the exclusive message queue.

func New

func New(redisURL string) *Red

New creates a named instance of the exclusive queue

func (*Red) Dequeue

func (r *Red) Dequeue(ctx context.Context) (*Message, error)

Dequeue message from the queue

func (*Red) Enqueue

func (r *Red) Enqueue(ctx context.Context, key, value interface{}) error

Enqueue message into queue

func (Red) Error

func (r Red) Error() error

Error checks if any errors have occurred

func (Red) Lock

func (r Red) Lock(name string, opts ...redsync.Option) *redsync.Mutex

Lock attempts to obtain a named lock for read/write

func (Red) Once

func (r Red) Once(key string)

Once schedules a task to be done once

func (Red) RLock

func (r Red) RLock(name string, opts ...redsync.Option) *redsync.Mutex

RLock attempts to obtain a named lock for reading

func (Red) Repeat

func (r Red) Repeat(key, recurrence string) error

Repeat schedules a task to be done at least once

func (Red) StartScheduling

func (r Red) StartScheduling(handler func(task *Task), schedulers ...func())

StartScheduling tasks

func (Red) StopScheduling

func (r Red) StopScheduling()

StopScheduling tasks

func (Red) Wait

func (r Red) Wait()

Wait for next task to be handled

type Scheduler

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

Scheduler is an instance of a persistent background scheduler

func NewScheduler

func NewScheduler(queue *Red) *Scheduler

NewScheduler creates a scheduler instance.

func (*Scheduler) Add

func (s *Scheduler) Add(tasks ...*Task) *Scheduler

Add adds a task to be scheduled.

func (*Scheduler) DequeueTimeout

func (s *Scheduler) DequeueTimeout(timeout time.Duration) *Scheduler

DequeueTimeout sets the maximum time to wait for remove items from queue.

func (*Scheduler) EnqueueTimeout

func (s *Scheduler) EnqueueTimeout(timeout time.Duration) *Scheduler

EnqueueTimeout sets the maximum time to wait for adding items to queue.

func (*Scheduler) Every

func (s *Scheduler) Every(interval time.Duration) *Scheduler

Every sets the interval for checking for new jobs to scheduler.

func (*Scheduler) Handle

func (s *Scheduler) Handle(fn func(task *Task))

Handle tasks when scheduled

func (*Scheduler) NewTask

func (s *Scheduler) NewTask(id string) *Task

NewTask creates a new instance of a task to be scheduled.

func (*Scheduler) Start

func (s *Scheduler) Start()

Start begins scheduling tasks.

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop stops the scheduler and waits for background jobs to finish.

type Task

type Task struct {
	Id       string       `json:"id"`
	Schedule TaskSchedule `json:"schedule"`
	*redsync.Mutex
}

Task is an instance of a thing to be scheduled.

func (*Task) CanSchedule

func (t *Task) CanSchedule(now time.Time) bool

CanSchedule determines if the task can be scheduled at given time.

func (*Task) IsComplete

func (t *Task) IsComplete() bool

IsComplete checks if the tasks can no longer be scheduled.

func (*Task) MarkScheduled

func (t *Task) MarkScheduled(at time.Time) *Task

MarkScheduled marks the task as scheduled.

func (*Task) Occurrences

func (t *Task) Occurrences() int

Occurrences gets the number of times the task has been scheduled.

func (*Task) SetRecurrence

func (t *Task) SetRecurrence(rfc string) error

SetRecurrence sets a new recurrence rule based on rfc 5545

type TaskSchedule

type TaskSchedule struct {
	Recurrence string    `json:"recurrence"`
	Count      int       `json:"count"`
	UpdatedAt  time.Time `json:"updatedAt"`
	sync.RWMutex
}

TaskSchedule is the schedule for when the task is to occur.

type Worker

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

Worker is an instance of a background worker.

func NewQWorker

func NewQWorker(name string, jobs ...func()) *Worker

NewQWorker creates a new worker instance with quiet mode enabled.

func NewWorker

func NewWorker(name string, jobs ...func()) *Worker

NewWorker creates a new worker instance.

func (*Worker) AddJobs

func (w *Worker) AddJobs(jobs ...func())

AddJobs adds jobs to the worker

func (*Worker) Concurrent

func (w *Worker) Concurrent(instances int) *Worker

Concurrent sets the number of simultaneous instances to process tasks.

func (*Worker) Every

func (w *Worker) Every(interval time.Duration) *Worker

Every sets the time between processing tasks.

func (*Worker) Start

func (w *Worker) Start()

Start begins processing tasks.

func (*Worker) Stop

func (w *Worker) Stop()

Stop stops the worker and waits for all instances to terminate.