tasqueue

package module
v1.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 5, 2023 License: BSD-2-Clause Imports: 13 Imported by: 0

README

taskqueue

Run Tests Go Report Card

Tasqueue is a simple, lightweight distributed job/worker implementation in Go

Installation

go get -u github.com/kalbhor/tasqueue

Concepts

  • tasqueue.Broker is a generic interface to enqueue and consume messages from a single queue. Currently supported brokers are redis and nats-jetstream. Note: It is important for the broker (or your enqueue, consume implementation) to guarantee atomicity. ie : Tasqueue does not provide locking capabilities to ensure unique job consumption.
  • tasqueue.Results is a generic interface to store the status and results of jobs. Currently supported result stores are redis and nats-jetstream.
  • tasqueue.Task is a pre-registered job handler. It stores a handler functions which is called to process a job. It also stores callbacks (if set through options), executed during different states of a job.
  • tasqueue.Job represents a unit of work pushed to a queue for consumption. It holds:
    • []byte payload (encoded in any manner, if required)
    • task name used to identify the pre-registed task which will processes the job.

Server

A tasqueue server is the main store that holds the broker and the results interfaces. It also acts as a hub to register tasks.

Server Options

Server options are used to configure the server. Broker & Results are mandatory, while logger and open telemetry provider are optional. Refer to the in-memory example for an open telemetry implementation.

type ServerOpts struct {
	// Mandatory results & broker implementations.
	Broker        Broker
	Results       Results

	// Optional logger and telemetry provider.
	Logger        logf.Logger
	TraceProvider *trace.TracerProvider
}

Usage

package main

import (
	"log"

	"github.com/kalbhor/tasqueue"
	rb "github.com/kalbhor/tasqueue/brokers/redis"
	rr "github.com/kalbhor/tasqueue/results/redis"
	"github.com/zerodha/logf"
)

func main() {
	lo := logf.New(logf.Opts{})

	broker := rb.New(rb.Options{
		Addrs:    []string{"127.0.0.1:6379"},
		Password: "",
		DB:       0,
	}, lo)
	results := rr.New(rr.Options{
		Addrs:    []string{"127.0.0.1:6379"},
		Password: "",
		DB:       0,
	}, lo)

	srv, err := tasqueue.NewServer(tasqueue.ServerOpts{
		Broker:        broker,
		Results:       results,
		Logger:        lo,
	})
	if err != nil {
		log.Fatal(err)
	}
}

Task Options

Concurrency is the number of processors run for this task. Queue is the queue to consume for this task. Task options contains callbacks that are executed one a state change.

type TaskOpts struct {
	Concurrency  uint32
	Queue        string
	SuccessCB    func(JobCtx)
	ProcessingCB func(JobCtx)
	RetryingCB   func(JobCtx)
	FailedCB     func(JobCtx)
}

Registering tasks

A task can be registered by supplying a name, handler and options. Jobs can be processed using a task registered by a particular name. A handler is a function with the signature func([]byte, JobCtx) error. It is the responsibility of the handler to deal with the []byte payload in whatever manner (decode, if required).

package tasks

import (
	"encoding/json"

	"github.com/kalbhor/tasqueue"
)

type SumPayload struct {
	Arg1 int `json:"arg1"`
	Arg2 int `json:"arg2"`
}

type SumResult struct {
	Result int `json:"result"`
}

// SumProcessor prints the sum of two integer arguements.
func SumProcessor(b []byte, m tasqueue.JobCtx) error {
	var pl SumPayload
	if err := json.Unmarshal(b, &pl); err != nil {
		return err
	}

	rs, err := json.Marshal(SumResult{Result: pl.Arg1 + pl.Arg2})
	if err != nil {
		return err
	}

	m.Save(rs)

	return nil
}

srv.RegisterTask("add", tasks.SumProcessor, TaskOpts{Concurrency: 5})

Start server

Start() starts the job consumer and processor. It is a blocking function. It listens for jobs on the queue and spawns processor go routines.

srv.Start(ctx)

Job

A tasqueue job represents a unit of work pushed onto the queue, that requires processing using a registered Task. It holds a []byte payload, a task name (which will process the payload) and various options.

Job Options

// JobOpts holds the various options available to configure a job.
type JobOpts struct {
	Queue      string // default: `tasqueue:tasks`
	MaxRetries uint32 // default: `1`
	Schedule   string // cron schedule for the job
}

Creating a job

NewJob returns a job with the supplied payload. It accepts the name of the task, the payload and a list of options.

b, _ := json.Marshal(tasks.SumPayload{Arg1: 5, Arg2: 4})
job, err := tasqueue.NewJob("add", b, tasqueue.JobOpts{})
if err != nil {
	log.Fatal(err)
}

Enqueuing a job

Once a job is created, it can be enqueued via the server for processing. Calling srv.Enqueue returns a job uuid which can be used to query the status of the job.

uuid, err := srv.Enqueue(ctx, job)
if err != nil {
	log.Fatal(err)
}

Getting a job message

To query the details of a job that was enqueued, we can use srv.GetJob. It returns a JobMessage which contains details related to a job.

jobMsg, err := srv.GetJob(ctx, uuid)
if err != nil {
	log.Fatal(err)
}

Fields available in a JobMessage (embeds Meta):

// Meta contains fields related to a job. These are updated when a task is consumed.
type Meta struct {
	UUID        string
	Status      string
	Queue       string
	Schedule    string
	MaxRetry    uint32
	Retried     uint32
	PrevErr     string
	ProcessedAt time.Time
}

JobCtx

JobCtx is passed to handler functions and callbacks. It can be used to view the job's meta information (JobCtx embeds Meta) and also to save arbitrary results for a job using func (c *JobCtx) Save(b []byte) error

Group

A tasqueue group holds multiple jobs and pushes them all simultaneously onto the queue, the Group is considered successful only if all the jobs finish successfully.

Creating a group

NewGroup returns a Group holding the jobs passed.

var group []tasqueue.Job

for i := 0; i < 3; i++ {
	b, _ := json.Marshal(tasks.SumPayload{Arg1: i, Arg2: 4})
	job, err := tasqueue.NewJob("add", b)
	if err != nil {
			log.Fatal(err)
	}
	group = append(group, job)
}

grp, err := tasqueue.NewGroup(group...)
if err != nil {
	log.Fatal(err)
}

Enqueuing a group

Once a group is created, it can be enqueued via the server for processing. Calling srv.EnqueueGroup returns a group uuid which can be used to query the status of the group.

groupUUID, err := srv.EnqueueGroup(ctx, grp)
if err != nil {
	log.Fatal(err)
}

Getting a group message

To query the details of a group that was enqueued, we can use srv.GetGroup. It returns a GroupMessage which contains details related to a group.

groupMsg, err := srv.GetGroup(ctx, groupUUID)
if err != nil {
	log.Fatal(err)
}

Fields available in a GroupMessage (embeds GroupMeta):

// GroupMeta contains fields related to a group job. These are updated when a task is consumed.
type GroupMeta struct {
	UUID   string
	Status string
	// JobStatus is a map of individual job uuid -> status
	JobStatus map[string]string
}

Chain

A tasqueue chain holds multiple jobs and pushes them one after the other (after a job succeeds), the Chain is considered successful only if the final job completes successfuly.

Creating a chain

NewChain returns a chain holding the jobs passed in the order.

var chain []tasqueue.Job

for i := 0; i < 3; i++ {
	b, _ := json.Marshal(tasks.SumPayload{Arg1: i, Arg2: 4})
	task, err := tasqueue.NewJob("add", b)
	if err != nil {
		log.Fatal(err)
	}
	chain = append(chain, task)
}

chn, err := tasqueue.NewChain(chain...)
if err != nil {
	log.Fatal(err)
}

Enqueuing a chain

Once a chain is created, it can be enqueued via the server for processing. Calling srv.EnqueueChain returns a chain uuid which can be used to query the status of the chain.

chainUUID, err := srv.EnqueueChain(ctx, chn)
if err != nil {
	log.Fatal(err)
}

Getting results of previous job in a chain

A job in the chain can access the results of the previous job in the chain by getting JobCtx.Meta.PrevJobResults. This will contain any job result saved by the previous job by JobCtx.Save().

Getting a chain message

To query the details of a chain that was enqueued, we can use srv.GetChain. It returns a ChainMessage which contains details related to a chian.

chainMsg, err := srv.GetChain(ctx, chainUUID)
if err != nil {
	log.Fatal(err)
}

Fields available in a ChainMessage (embeds ChainMeta):

// ChainMeta contains fields related to a chain job.
type ChainMeta struct {
	UUID string
	// Status of the overall chain
	Status string
	// UUID of the current job part of chain
	JobUUID string
	// List of UUIDs of completed jobs
	PrevJobs []string
}

Result

A result is arbitrary []byte data saved by a handler or callback via JobCtx.Save().

Get Result

b, err := srv.GetResult(ctx, jobUUID)
if err != nil {
	log.Fatal(err)
}

Credits

  • @knadh for the logo & feature suggestions

License

BSD-2-Clause-FreeBSD

Documentation

Index

Constants

View Source 
const (

	// This is the initial state when a job is pushed onto the broker.
	StatusStarted = "queued"

	// This is the state when a worker has recieved a job.
	StatusProcessing = "processing"

	// The state when a job completes, but returns an error (and all retries are over).
	StatusFailed = "failed"

	// The state when a job completes without any error.
	StatusDone = "successful"

	// The state when a job errors out and is queued again to be retried.
	// This state is analogous to statusStarted.
	StatusRetrying = "retrying"
)
View Source 
const (
	DefaultQueue = "tasqueue:tasks"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type Broker 

type Broker interface {
	// Enqueue places a task in the queue
	Enqueue(ctx context.Context, msg []byte, queue string) error

	// Consume listens for tasks on the queue and calls processor
	Consume(ctx context.Context, work chan []byte, queue string)

	// GetPending returns a list of stored job messages on the particular queue
	GetPending(ctx context.Context, queue string) ([]string, error)
}

type Chain 

type Chain struct {
	Jobs []Job
}

func NewChain 

func NewChain(j ...Job) (Chain, error)

NewChain() accepts a list of Tasks and creates a chain by setting the onSuccess task of i'th task to (i+1)'th task, hence forming a "chain". It returns the first task (essentially the first node of the linked list), which can be queued normally.

type ChainMessage 

type ChainMessage struct {
	ChainMeta
}

ChainMessage is a wrapper over Chain, containing meta info such as status, uuid. A ChainMessage is stored in the results store.

type ChainMeta 

type ChainMeta struct {
	UUID string
	// Status of the overall chain
	Status string
	// UUID of the current job part of chain
	JobUUID string
	// List of UUIDs of completed jobs
	PrevJobs []string
}

ChainMeta contains fields related to a chain job.

type Group 

type Group struct {
	Jobs []Job
}

func NewGroup 

func NewGroup(j ...Job) (Group, error)

NewGroup() accepts a list of jobs and creates a group.

type GroupMessage 

type GroupMessage struct {
	GroupMeta
	Group *Group
}

GroupMessage is a wrapper over Group, containing meta info such as status, uuid. A GroupMessage is stored in the results store.

type GroupMeta 

type GroupMeta struct {
	UUID   string
	Status string
	// JobStatus is a map of job uuid -> status
	JobStatus map[string]string
}

GroupMeta contains fields related to a group job. These are updated when a task is consumed.

type Job 

type Job struct {
	// If task is successful, the OnSuccess jobs are enqueued.
	OnSuccess *Job
	Task      string
	Payload   []byte

	Opts JobOpts
}

Job represents a unit of work pushed by producers. It is the responsibility of the task handler to unmarshal (if required) the payload and process it in any manner.

func NewJob 

func NewJob(handler string, payload []byte, opts JobOpts) (Job, error)

NewJob returns a job with arbitrary payload. It accepts the name of the task, the payload and a list of options.

type JobCtx 

type JobCtx struct {
	context.Context

	Meta Meta
	// contains filtered or unexported fields
}

JobCtx is passed onto handler functions. It allows access to a job's meta information to the handler.

func (JobCtx) Save 

func (c JobCtx) Save(b []byte) error

Save() sets arbitrary results for a job in the results store.

type JobMessage 

type JobMessage struct {
	Meta
	Job *Job
}

JobMessage is a wrapper over Task, used to transport the task over a broker. It contains additional fields such as status and a UUID.

type JobOpts 

type JobOpts struct {
	Queue      string
	MaxRetries uint32
	Schedule   string
	Timeout    time.Duration
}

JobOpts holds the various options available to configure a job.

type Meta 

type Meta struct {
	UUID          string
	OnSuccessUUID string
	Status        string
	Queue         string
	Schedule      string
	MaxRetry      uint32
	Retried       uint32
	PrevErr       string
	ProcessedAt   time.Time

	// PrevJobResults contains any job result set by the previous job in a chain.
	// This will be nil if the previous job doesn't set the results on JobCtx.
	PrevJobResult []byte
}

Meta contains fields related to a job. These are updated when a task is consumed.

func DefaultMeta added in v0.2.0 

func DefaultMeta(opts JobOpts) Meta

DefaultMeta returns Meta with a UUID and other defaults filled in.

type Results 

type Results interface {
	Get(ctx context.Context, uuid string) ([]byte, error)
	Set(ctx context.Context, uuid string, b []byte) error
	// DeleteJob removes the job's saved metadata from the store
	DeleteJob(ctx context.Context, uuid string) error
	GetFailed(ctx context.Context) ([]string, error)
	GetSuccess(ctx context.Context) ([]string, error)
	SetFailed(ctx context.Context, uuid string) error
	SetSuccess(ctx context.Context, uuid string) error
}

type Server 

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

Server is the main store that holds the broker and the results communication interfaces. It also stores the registered tasks.

func NewServer 

func NewServer(o ServerOpts) (*Server, error)

NewServer() returns a new instance of server, with sane defaults.

func (*Server) DeleteJob added in v1.3.0 

func (s *Server) DeleteJob(ctx context.Context, uuid string) error

DeleteJob() removes the stored results of a particular job. It does not "dequeue" an unprocessed job. It is useful for removing the status of old finished jobs.

func (*Server) Enqueue 

func (s *Server) Enqueue(ctx context.Context, t Job) (string, error)

Enqueue() accepts a job and returns the assigned UUID. The following steps take place: 1. Converts it into a job message, which assigns a UUID (among other meta info) to the job. 2. Sets the job status as "started" on the results store. 3. Enqueues the job (if the job is scheduled, pushes it onto the scheduler)

func (*Server) EnqueueChain 

func (s *Server) EnqueueChain(ctx context.Context, c Chain) (string, error)

func (*Server) EnqueueGroup 

func (s *Server) EnqueueGroup(ctx context.Context, t Group) (string, error)

EnqueueGroup() accepts a group and returns the assigned UUID. The following steps take place: 1. Converts it into a group message, which assigns a UUID (among other meta info) to the group. 2. Sets the group status as "started" on the results store. 3. Loops over all jobs part of the group and enqueues the job each job. 4. The job status map is updated with the uuids of each enqueued job.

func (*Server) GetChain 

func (s *Server) GetChain(ctx context.Context, uuid string) (ChainMessage, error)

func (*Server) GetFailed added in v0.3.0 

func (s *Server) GetFailed(ctx context.Context) ([]string, error)

GetFailed() returns the list of uuid's of jobs that failed.

func (*Server) GetGroup 

func (s *Server) GetGroup(ctx context.Context, uuid string) (GroupMessage, error)

func (*Server) GetJob 

func (s *Server) GetJob(ctx context.Context, uuid string) (JobMessage, error)

GetJob accepts a UUID and returns the job message in the results store. This is useful to check the status of a job message.

func (*Server) GetPending added in v1.3.0 

func (s *Server) GetPending(ctx context.Context, queue string) ([]JobMessage, error)

GetPending() returns the pending job message's in the broker's queue.

func (*Server) GetResult 

func (s *Server) GetResult(ctx context.Context, uuid string) ([]byte, error)

GetResult() accepts a UUID and returns the result of the job in the results store.

func (*Server) GetSuccess added in v0.3.0 

func (s *Server) GetSuccess(ctx context.Context) ([]string, error)

GetSuccess() returns the list of uuid's of jobs that were successful.

func (*Server) GetTasks added in v1.3.0 

func (s *Server) GetTasks() []string

GetTasks() returns a list of all tasks registered with the server.

func (*Server) RegisterTask 

func (s *Server) RegisterTask(name string, fn handler, opts TaskOpts)

RegisterTask maps a new task against the tasks map on the server. It accepts different options for the task (to set callbacks).

func (*Server) Start 

func (s *Server) Start(ctx context.Context)

Start() starts the job consumer and processor. It is a blocking function.

type ServerOpts 

type ServerOpts struct {
	Broker        Broker
	Results       Results
	Logger        logf.Logger
	TraceProvider *trace.TracerProvider
}

type Task 

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

Task is a pre-registered job handler. It stores the callbacks (set through options), which are called during different states of a job.

type TaskOpts 

type TaskOpts struct {
	Concurrency  uint32
	Queue        string
	SuccessCB    func(JobCtx)
	ProcessingCB func(JobCtx)
	RetryingCB   func(JobCtx)
	FailedCB     func(JobCtx)
}

Source Files

View all Source files

Directories

Path Synopsis
brokers
in-memory
nats-js
redis
examples
in-memory
nats-js
redis
tasks
results
in-memory
nats-js
redis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.