tasqueue

package module
v2.2.0 Latest Latest
Warning

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

 Go to latest Published: Mar 14, 2024 License: BSD-2-Clause Imports: 14 Imported by: 1

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/v2

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/v2"
	rb "github.com/kalbhor/tasqueue/v2/brokers/redis"
	rr "github.com/kalbhor/tasqueue/v2/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

Queue is the name of the queue assigned to the task. By default the value is "tasqueue:tasks". Queues can be shared between tasks.

Concurrency defines the number of processor go-routines running on the queue. By default this number is equal to runtime.GOMAXPROCS(0) (number of CPUs on the system). Ideally, it is recommended that the client tweak this number according to their tasks.

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/v2"
)

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
}

Once a queue is created if the client creates a task with an existing queue but supplies a different concurrency in the TaskOpts, then RegisterTask will return an error.

// This creates the q1 queue if it doesn't exist and assigns 5 concurrency to it
err := srv.RegisterTask("add", tasks.SumProcessor, TaskOpts{Queue:"q1", Concurrency: 5})
if err != nil {
	log.Fatal(err)
}

// No error
err := srv.RegisterTask("div", tasks.DivProcessor, TaskOpts{Queue:"q1"})
if err != nil {
	log.Fatal(err)
}

// No error
err := srv.RegisterTask("sub", tasks.SubProcessor, TaskOpts{Queue:"q1", Concurrency: 5})
if err != nil {
	log.Fatal(err)
}

// This will return an error since q1 is already created and its concurrency cannot be modified
err := srv.RegisterTask("multiplication", tasks.MulProcessor, TaskOpts{Queue:"q1", Concurrency: 10})
if err != nil {
	log.Fatal(err)
}

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 {
	// Optional ID passed by client. If empty, Tasqueue generates it.
	ID string

	Queue      string
	MaxRetries uint32
	Schedule   string
	Timeout    time.Duration
}

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 id which can be used to query the status of the job.

id, 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, id)
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 {
	ID          string
	OnSuccessID 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
}

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, tasqueue.GroupOpts{})
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 id which can be used to query the status of the group.

groupID, 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, groupID)
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 {
	ID   string
	Status string
	// JobStatus is a map of individual job id -> 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, tasqueue.ChainOpts{})
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 id which can be used to query the status of the chain.

chainID, 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, chainID)
if err != nil {
	log.Fatal(err)
}

Fields available in a ChainMessage (embeds ChainMeta):

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

Result

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

Get Result

If the jobID does not exist, ErrNotFound will be returned

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

Delete Result

DeleteJob removes the job's saved metadata from the store

err := srv.DeleteResult(ctx, jobID)
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

View Source 
var ErrNotFound = errors.New("result not found")

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

	// EnqueueScheduled accepts a task (msg, queue) and also a timestamp
	// The job should be enqueued at the particular timestamp.
	EnqueueScheduled(ctx context.Context, msg []byte, queue string, ts time.Time) 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
	Opts ChainOpts
}

func NewChain 

func NewChain(j []Job, opts ChainOpts) (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, id. A ChainMessage is stored in the results store.

type ChainMeta 

type ChainMeta struct {
	ID string
	// Status of the overall chain
	Status string
	// ID of the current job part of chain
	JobID string
	// List of IDs of completed jobs
	PrevJobs []string
}

ChainMeta contains fields related to a chain job.

type ChainOpts 

type ChainOpts struct {
	// Optional ID passed by client. If empty, Tasqueue generates it.
	ID string
}

type Group 

type Group struct {
	Jobs []Job
	Opts GroupOpts
}

func NewGroup 

func NewGroup(j []Job, opts GroupOpts) (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, id. A GroupMessage is stored in the results store.

type GroupMeta 

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

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

type GroupOpts 

type GroupOpts struct {
	ID string
}

type Job 

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

	// If task fails, the OnError jobs are enqueued.
	OnError []*Job

	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 ID.

type JobOpts 

type JobOpts struct {
	// Optional ID passed by client. If empty, Tasqueue generates it.
	ID string

	ETA        time.Time
	Queue      string
	MaxRetries uint32
	Schedule   string
	Timeout    time.Duration
}

JobOpts holds the various options available to configure a job.

type Meta 

type Meta struct {
	ID           string
	OnSuccessIDs []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 

func DefaultMeta(opts JobOpts) Meta

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

type Results 

type Results interface {
	Get(ctx context.Context, id string) ([]byte, error)
	// NilError is used to check internally if the "id" is missing
	NilError() error
	Set(ctx context.Context, id string, b []byte) error
	// DeleteJob removes the job's saved metadata from the store
	DeleteJob(ctx context.Context, id string) error
	GetFailed(ctx context.Context) ([]string, error)
	GetSuccess(ctx context.Context) ([]string, error)
	SetFailed(ctx context.Context, id string) error
	SetSuccess(ctx context.Context, id 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 

func (s *Server) DeleteJob(ctx context.Context, id 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 ID. The following steps take place: 1. Converts it into a job message, which assigns a ID (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 ID. The following steps take place: 1. Converts it into a group message, which assigns a ID (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 IDs of each enqueued job.

func (*Server) GetChain 

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

func (*Server) GetFailed 

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

GetFailed() returns the list of ids of jobs that failed.

func (*Server) GetGroup 

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

func (*Server) GetJob 

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

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

func (*Server) GetPending 

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, id string) ([]byte, error)

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

func (*Server) GetSuccess 

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

GetSuccess() returns the list of ids of jobs that were successful.

func (*Server) GetTasks 

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) error

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        slog.Handler
	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, error)
	FailedCB     func(JobCtx, error)
}

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.