asynq

package module
v0.25.2 Latest Latest
Warning

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

 Go to latest Published: Dec 16, 2024 License: MIT Imports: 29 Imported by: 2

README

Asynq logo

Simple, reliable & efficient distributed task queue in Go

GoDoc Go Report Card Build Status Gitter chat

Asynq is a Go library for queueing tasks and processing them asynchronously with workers. It's backed by Redis and is designed to be scalable yet easy to get started.

Highlevel overview of how Asynq works:

  • Client puts tasks on a queue
  • Server pulls tasks off queues and starts a worker goroutine for each task
  • Tasks are processed concurrently by multiple workers

Task queues are used as a mechanism to distribute work across multiple machines. A system can consist of multiple worker servers and brokers, giving way to high availability and horizontal scaling.

Example use case

Task Queue Diagram

Features

Stability and Compatibility

Status: The library relatively stable and is currently undergoing moderate development with less frequent breaking API changes.

☝️ Important Note: Current major version is zero (v0.x.x) to accommodate rapid development and fast iteration while getting early feedback from users (feedback on APIs are appreciated!). The public API could change without a major version update before v1.0.0 release.

Redis Cluster Compatibility

Some of the lua scripts in this library may not be compatible with Redis Cluster.

Sponsoring

If you are using this package in production, please consider sponsoring the project to show your support!

Quickstart

Make sure you have Go installed (download). The last two Go versions are supported (See https://go.dev/dl).

Initialize your project by creating a folder and then running go mod init github.com/your/repo (learn more) inside the folder. Then install Asynq library with the go get command:

go get -u github.com/OlegBugaichuk/asynq

Make sure you're running a Redis server locally or from a Docker container. Version 4.0 or higher is required.

Next, write a package that encapsulates task creation and task handling.

package tasks

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"
    "github.com/OlegBugaichuk/asynq"
)

// A list of task types.
const (
    TypeEmailDelivery   = "email:deliver"
    TypeImageResize     = "image:resize"
)

type EmailDeliveryPayload struct {
    UserID     int
    TemplateID string
}

type ImageResizePayload struct {
    SourceURL string
}

//----------------------------------------------
// Write a function NewXXXTask to create a task.
// A task consists of a type and a payload.
//----------------------------------------------

func NewEmailDeliveryTask(userID int, tmplID string) (*asynq.Task, error) {
    payload, err := json.Marshal(EmailDeliveryPayload{UserID: userID, TemplateID: tmplID})
    if err != nil {
        return nil, err
    }
    return asynq.NewTask(TypeEmailDelivery, payload), nil
}

func NewImageResizeTask(src string) (*asynq.Task, error) {
    payload, err := json.Marshal(ImageResizePayload{SourceURL: src})
    if err != nil {
        return nil, err
    }
    // task options can be passed to NewTask, which can be overridden at enqueue time.
    return asynq.NewTask(TypeImageResize, payload, asynq.MaxRetry(5), asynq.Timeout(20 * time.Minute)), nil
}

//---------------------------------------------------------------
// Write a function HandleXXXTask to handle the input task.
// Note that it satisfies the asynq.HandlerFunc interface.
//
// Handler doesn't need to be a function. You can define a type
// that satisfies asynq.Handler interface. See examples below.
//---------------------------------------------------------------

func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    var p EmailDeliveryPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Sending Email to User: user_id=%d, template_id=%s", p.UserID, p.TemplateID)
    // Email delivery code ...
    return nil
}

// ImageProcessor implements asynq.Handler interface.
type ImageProcessor struct {
    // ... fields for struct
}

func (processor *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    var p ImageResizePayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Resizing image: src=%s", p.SourceURL)
    // Image resizing code ...
    return nil
}

func NewImageProcessor() *ImageProcessor {
	return &ImageProcessor{}
}

In your application code, import the above package and use Client to put tasks on queues.

package main

import (
    "log"
    "time"

    "github.com/OlegBugaichuk/asynq"
    "your/app/package/tasks"
)

const redisAddr = "127.0.0.1:6379"

func main() {
    client := asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddr})
    defer client.Close()

    // ------------------------------------------------------
    // Example 1: Enqueue task to be processed immediately.
    //            Use (*Client).Enqueue method.
    // ------------------------------------------------------

    task, err := tasks.NewEmailDeliveryTask(42, "some:template:id")
    if err != nil {
        log.Fatalf("could not create task: %v", err)
    }
    info, err := client.Enqueue(task)
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)


    // ------------------------------------------------------------
    // Example 2: Schedule task to be processed in the future.
    //            Use ProcessIn or ProcessAt option.
    // ------------------------------------------------------------

    info, err = client.Enqueue(task, asynq.ProcessIn(24*time.Hour))
    if err != nil {
        log.Fatalf("could not schedule task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)


    // ----------------------------------------------------------------------------
    // Example 3: Set other options to tune task processing behavior.
    //            Options include MaxRetry, Queue, Timeout, Deadline, Unique etc.
    // ----------------------------------------------------------------------------

    task, err = tasks.NewImageResizeTask("https://example.com/myassets/image.jpg")
    if err != nil {
        log.Fatalf("could not create task: %v", err)
    }
    info, err = client.Enqueue(task, asynq.MaxRetry(10), asynq.Timeout(3 * time.Minute))
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
}

Next, start a worker server to process these tasks in the background. To start the background workers, use Server and provide your Handler to process the tasks.

You can optionally use ServeMux to create a handler, just as you would with net/http Handler.

package main

import (
    "log"

    "github.com/OlegBugaichuk/asynq"
    "your/app/package/tasks"
)

const redisAddr = "127.0.0.1:6379"

func main() {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: redisAddr},
        asynq.Config{
            // Specify how many concurrent workers to use
            Concurrency: 10,
            // Optionally specify multiple queues with different priority.
            Queues: map[string]int{
                "critical": 6,
                "default":  3,
                "low":      1,
            },
            // See the godoc for other configuration options
        },
    )

    // mux maps a type to a handler
    mux := asynq.NewServeMux()
    mux.HandleFunc(tasks.TypeEmailDelivery, tasks.HandleEmailDeliveryTask)
    mux.Handle(tasks.TypeImageResize, tasks.NewImageProcessor())
    // ...register other handlers...

    if err := srv.Run(mux); err != nil {
        log.Fatalf("could not run server: %v", err)
    }
}

For a more detailed walk-through of the library, see our Getting Started guide.

To learn more about asynq features and APIs, see the package godoc.

Web UI

Asynqmon is a web based tool for monitoring and administrating Asynq queues and tasks.

Here's a few screenshots of the Web UI:

Queues view

Web UI Queues View

Tasks view

Web UI TasksView

Metrics view Screen Shot 2021-12-19 at 4 37 19 PM

Settings and adaptive dark mode

Web UI Settings and adaptive dark mode

For details on how to use the tool, refer to the tool's README.

Command Line Tool

Asynq ships with a command line tool to inspect the state of queues and tasks.

To install the CLI tool, run the following command:

go install github.com/OlegBugaichuk/asynq/tools/asynq@latest

Here's an example of running the asynq dash command:

Gif

For details on how to use the tool, refer to the tool's README.

Contributing

We are open to, and grateful for, any contributions (GitHub issues/PRs, feedback on Gitter channel, etc) made by the community.

Please see the Contribution Guide before contributing.

License

Copyright (c) 2019-present Ken Hibino and Contributors. Asynq is free and open-source software licensed under the MIT License. Official logo was created by Vic Shóstak and distributed under Creative Commons license (CC0 1.0 Universal).

Documentation

Overview

Package asynq provides a framework for Redis based distrubted task queue.

Asynq uses Redis as a message broker. To connect to redis, specify the connection using one of RedisConnOpt types. 

redisConnOpt = asynq.RedisClientOpt{
    Addr:     "127.0.0.1:6379",
    Password: "xxxxx",
    DB:       2,
}

The Client is used to enqueue a task. 

client := asynq.NewClient(redisConnOpt)

// Task is created with two parameters: its type and payload.
// Payload data is simply an array of bytes. It can be encoded in JSON, Protocol Buffer, Gob, etc.
b, err := json.Marshal(ExamplePayload{UserID: 42})
if err != nil {
    log.Fatal(err)
}

task := asynq.NewTask("example", b)

// Enqueue the task to be processed immediately.
info, err := client.Enqueue(task)

// Schedule the task to be processed after one minute.
info, err = client.Enqueue(t, asynq.ProcessIn(1*time.Minute))

The Server is used to run the task processing workers with a given handler. 

srv := asynq.NewServer(redisConnOpt, asynq.Config{
    Concurrency: 10,
})

if err := srv.Run(handler); err != nil {
    log.Fatal(err)
}

Handler is an interface type with a method which takes a task and returns an error. Handler should return nil if the processing is successful, otherwise return a non-nil error. If handler panics or returns a non-nil error, the task will be retried in the future.

Example of a type that implements the Handler interface. 

type TaskHandler struct {
    // ...
}

func (h *TaskHandler) ProcessTask(ctx context.Context, task *asynq.Task) error {
    switch task.Type {
    case "example":
        var data ExamplePayload
        if err := json.Unmarshal(task.Payload(), &data); err != nil {
            return err
        }
        // perform task with the data

    default:
        return fmt.Errorf("unexpected task type %q", task.Type)
    }
    return nil
}

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// ErrQueueNotFound indicates that the specified queue does not exist.
	ErrQueueNotFound = errors.New("queue not found")

	// ErrQueueNotEmpty indicates that the specified queue is not empty.
	ErrQueueNotEmpty = errors.New("queue is not empty")

	// ErrTaskNotFound indicates that the specified task cannot be found in the queue.
	ErrTaskNotFound = errors.New("task not found")
)
View Source 
var ErrDuplicateTask = errors.New("task already exists")

ErrDuplicateTask indicates that the given task could not be enqueued since it's a duplicate of another task.

ErrDuplicateTask error only applies to tasks enqueued with a Unique option.

View Source 
var ErrLeaseExpired = errors.New("asynq: task lease expired")

ErrLeaseExpired error indicates that the task failed because the worker working on the task could not extend its lease due to missing heartbeats. The worker may have crashed or got cutoff from the network.

View Source 
var ErrServerClosed = errors.New("asynq: Server closed")

ErrServerClosed indicates that the operation is now illegal because of the server has been shutdown.

View Source 
var ErrTaskIDConflict = errors.New("task ID conflicts with another task")

ErrTaskIDConflict indicates that the given task could not be enqueued since its task ID already exists.

ErrTaskIDConflict error only applies to tasks enqueued with a TaskID option.

View Source 
var RevokeTask = errors.New("revoke task")

RevokeTask is used as a return value from Handler.ProcessTask to indicate that the task should not be retried or archived.

View Source 
var SkipRetry = errors.New("skip retry for the task")

SkipRetry is used as a return value from Handler.ProcessTask to indicate that the task should not be retried and should be archived instead.

Functions

func DefaultRetryDelayFunc 

func DefaultRetryDelayFunc(n int, e error, t *Task) time.Duration

DefaultRetryDelayFunc is the default RetryDelayFunc used if one is not specified in Config. It uses exponential back-off strategy to calculate the retry delay.

func GetMaxRetry 

func GetMaxRetry(ctx context.Context) (n int, ok bool)

GetMaxRetry extracts maximum retry from a context, if any.

Return value n indicates the maximum number of times the associated task can be retried if ProcessTask returns a non-nil error.

func GetQueueName 

func GetQueueName(ctx context.Context) (queue string, ok bool)

GetQueueName extracts queue name from a context, if any.

Return value queue indicates which queue the task was pulled from.

func GetRetryCount 

func GetRetryCount(ctx context.Context) (n int, ok bool)

GetRetryCount extracts retry count from a context, if any.

Return value n indicates the number of times associated task has been retried so far.

func GetTaskID 

func GetTaskID(ctx context.Context) (id string, ok bool)

GetTaskID extracts a task ID from a context, if any.

ID of a task is guaranteed to be unique. ID of a task doesn't change if the task is being retried.

func IsPanicError 

func IsPanicError(err error) bool

func NotFound 

func NotFound(ctx context.Context, task *Task) error

NotFound returns an error indicating that the handler was not found for the given task.

Types

type Client 

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

A Client is responsible for scheduling tasks.

A Client is used to register tasks that should be processed immediately or some time in the future.

Clients are safe for concurrent use by multiple goroutines.

func NewClient 

func NewClient(r RedisConnOpt) *Client

NewClient returns a new Client instance given a redis connection option.

func NewClientFromRedisClient 

func NewClientFromRedisClient(c redis.UniversalClient) *Client

NewClientFromRedisClient returns a new instance of Client given a redis.UniversalClient Warning: The underlying redis connection pool will not be closed by Asynq, you are responsible for closing it.

func (*Client) Close 

func (c *Client) Close() error

Close closes the connection with redis.

func (*Client) Enqueue 

func (c *Client) Enqueue(task *Task, opts ...Option) (*TaskInfo, error)

Enqueue enqueues the given task to a queue.

Enqueue returns TaskInfo and nil error if the task is enqueued successfully, otherwise returns a non-nil error.

The argument opts specifies the behavior of task processing. If there are conflicting Option values the last one overrides others. Any options provided to NewTask can be overridden by options passed to Enqueue. By default, max retry is set to 25 and timeout is set to 30 minutes.

If no ProcessAt or ProcessIn options are provided, the task will be pending immediately.

Enqueue uses context.Background internally; to specify the context, use EnqueueContext.

func (*Client) EnqueueContext 

func (c *Client) EnqueueContext(ctx context.Context, task *Task, opts ...Option) (*TaskInfo, error)

EnqueueContext enqueues the given task to a queue.

EnqueueContext returns TaskInfo and nil error if the task is enqueued successfully, otherwise returns a non-nil error.

The argument opts specifies the behavior of task processing. If there are conflicting Option values the last one overrides others. Any options provided to NewTask can be overridden by options passed to Enqueue. By default, max retry is set to 25 and timeout is set to 30 minutes.

If no ProcessAt or ProcessIn options are provided, the task will be pending immediately.

The first argument context applies to the enqueue operation. To specify task timeout and deadline, use Timeout and Deadline option instead.

func (*Client) Ping 

func (c *Client) Ping() error

Ping performs a ping against the redis connection.

type ClusterNode 

type ClusterNode struct {
	// Node ID in the cluster.
	ID string

	// Address of the node.
	Addr string
}

ClusterNode describes a node in redis cluster.

type Config 

type Config struct {
	// Maximum number of concurrent processing of tasks.
	//
	// If set to a zero or negative value, NewServer will overwrite the value
	// to the number of CPUs usable by the current process.
	Concurrency int

	// BaseContext optionally specifies a function that returns the base context for Handler invocations on this server.
	//
	// If BaseContext is nil, the default is context.Background().
	// If this is defined, then it MUST return a non-nil context
	BaseContext func() context.Context

	// TaskCheckInterval specifies the interval between checks for new tasks to process when all queues are empty.
	//
	// If unset, zero or a negative value, the interval is set to 1 second.
	//
	// Note: Setting this value too low may add significant load to redis.
	//
	// By default, TaskCheckInterval is set to 1 seconds.
	TaskCheckInterval time.Duration

	// Function to calculate retry delay for a failed task.
	//
	// By default, it uses exponential backoff algorithm to calculate the delay.
	RetryDelayFunc RetryDelayFunc

	// Predicate function to determine whether the error returned from Handler is a failure.
	// If the function returns false, Server will not increment the retried counter for the task,
	// and Server won't record the queue stats (processed and failed stats) to avoid skewing the error
	// rate of the queue.
	//
	// By default, if the given error is non-nil the function returns true.
	IsFailure func(error) bool

	// List of queues to process with given priority value. Keys are the names of the
	// queues and values are associated priority value.
	//
	// If set to nil or not specified, the server will process only the "default" queue.
	//
	// Priority is treated as follows to avoid starving low priority queues.
	//
	// Example:
	//
	//     Queues: map[string]int{
	//         "critical": 6,
	//         "default":  3,
	//         "low":      1,
	//     }
	//
	// With the above config and given that all queues are not empty, the tasks
	// in "critical", "default", "low" should be processed 60%, 30%, 10% of
	// the time respectively.
	//
	// If a queue has a zero or negative priority value, the queue will be ignored.
	Queues map[string]int

	// StrictPriority indicates whether the queue priority should be treated strictly.
	//
	// If set to true, tasks in the queue with the highest priority is processed first.
	// The tasks in lower priority queues are processed only when those queues with
	// higher priorities are empty.
	StrictPriority bool

	ErrorHandler ErrorHandler

	// Logger specifies the logger used by the server instance.
	//
	// If unset, default logger is used.
	Logger Logger

	// LogLevel specifies the minimum log level to enable.
	//
	// If unset, InfoLevel is used by default.
	LogLevel LogLevel

	// ShutdownTimeout specifies the duration to wait to let workers finish their tasks
	// before forcing them to abort when stopping the server.
	//
	// If unset or zero, default timeout of 8 seconds is used.
	ShutdownTimeout time.Duration

	// HealthCheckFunc is called periodically with any errors encountered during ping to the
	// connected redis server.
	HealthCheckFunc func(error)

	// HealthCheckInterval specifies the interval between healthchecks.
	//
	// If unset or zero, the interval is set to 15 seconds.
	HealthCheckInterval time.Duration

	// DelayedTaskCheckInterval specifies the interval between checks run on 'scheduled' and 'retry'
	// tasks, and forwarding them to 'pending' state if they are ready to be processed.
	//
	// If unset or zero, the interval is set to 5 seconds.
	DelayedTaskCheckInterval time.Duration

	// GroupGracePeriod specifies the amount of time the server will wait for an incoming task before aggregating
	// the tasks in a group. If an incoming task is received within this period, the server will wait for another
	// period of the same length, up to GroupMaxDelay if specified.
	//
	// If unset or zero, the grace period is set to 1 minute.
	// Minimum duration for GroupGracePeriod is 1 second. If value specified is less than a second, the call to
	// NewServer will panic.
	GroupGracePeriod time.Duration

	// GroupMaxDelay specifies the maximum amount of time the server will wait for incoming tasks before aggregating
	// the tasks in a group.
	//
	// If unset or zero, no delay limit is used.
	GroupMaxDelay time.Duration

	// GroupMaxSize specifies the maximum number of tasks that can be aggregated into a single task within a group.
	// If GroupMaxSize is reached, the server will aggregate the tasks into one immediately.
	//
	// If unset or zero, no size limit is used.
	GroupMaxSize int

	// GroupAggregator specifies the aggregation function used to aggregate multiple tasks in a group into one task.
	//
	// If unset or nil, the group aggregation feature will be disabled on the server.
	GroupAggregator GroupAggregator

	// JanitorInterval specifies the average interval of janitor checks for expired completed tasks.
	//
	// If unset or zero, default interval of 8 seconds is used.
	JanitorInterval time.Duration

	// JanitorBatchSize specifies the number of expired completed tasks to be deleted in one run.
	//
	// If unset or zero, default batch size of 100 is used.
	// Make sure to not put a big number as the batch size to prevent a long-running script.
	JanitorBatchSize int
}

Config specifies the server's background-task processing behavior.

type DailyStats 

type DailyStats struct {
	// Name of the queue.
	Queue string
	// Total number of tasks being processed during the given date.
	// The number includes both succeeded and failed tasks.
	Processed int
	// Total number of tasks failed to be processed during the given date.
	Failed int
	// Date this stats was taken.
	Date time.Time
}

DailyStats holds aggregate data for a given day for a given queue.

type ErrorHandler 

type ErrorHandler interface {
	HandleError(ctx context.Context, task *Task, err error)
}

An ErrorHandler handles an error occurred during task processing.

type ErrorHandlerFunc 

type ErrorHandlerFunc func(ctx context.Context, task *Task, err error)

The ErrorHandlerFunc type is an adapter to allow the use of ordinary functions as a ErrorHandler. If f is a function with the appropriate signature, ErrorHandlerFunc(f) is a ErrorHandler that calls f.

func (ErrorHandlerFunc) HandleError 

func (fn ErrorHandlerFunc) HandleError(ctx context.Context, task *Task, err error)

HandleError calls fn(ctx, task, err)

type GroupAggregator 

type GroupAggregator interface {
	// Aggregate aggregates the given tasks in a group with the given group name,
	// and returns a new task which is the aggregation of those tasks.
	//
	// Use NewTask(typename, payload, opts...) to set any options for the aggregated task.
	// The Queue option, if provided, will be ignored and the aggregated task will always be enqueued
	// to the same queue the group belonged.
	Aggregate(group string, tasks []*Task) *Task
}

GroupAggregator aggregates a group of tasks into one before the tasks are passed to the Handler.

type GroupAggregatorFunc 

type GroupAggregatorFunc func(group string, tasks []*Task) *Task

The GroupAggregatorFunc type is an adapter to allow the use of ordinary functions as a GroupAggregator. If f is a function with the appropriate signature, GroupAggregatorFunc(f) is a GroupAggregator that calls f.

func (GroupAggregatorFunc) Aggregate 

func (fn GroupAggregatorFunc) Aggregate(group string, tasks []*Task) *Task

Aggregate calls fn(group, tasks)

type GroupInfo 

type GroupInfo struct {
	// Name of the group.
	Group string

	// Size is the total number of tasks in the group.
	Size int
}

GroupInfo represents a state of a group at a certain time.

type Handler 

type Handler interface {
	ProcessTask(context.Context, *Task) error
}

A Handler processes tasks.

ProcessTask should return nil if the processing of a task is successful.

If ProcessTask returns a non-nil error or panics, the task will be retried after delay if retry-count is remaining, otherwise the task will be archived.

One exception to this rule is when ProcessTask returns a SkipRetry error. If the returned error is SkipRetry or an error wraps SkipRetry, retry is skipped and the task will be immediately archived instead.

Another exception to this rule is when ProcessTask returns a RevokeTask error. If the returned error is RevokeTask or an error wraps RevokeTask, the task will not be retried or archived.

func NotFoundHandler 

func NotFoundHandler() Handler

NotFoundHandler returns a simple task handler that returns a “not found“ error.

type HandlerFunc 

type HandlerFunc func(context.Context, *Task) error

The HandlerFunc type is an adapter to allow the use of ordinary functions as a Handler. If f is a function with the appropriate signature, HandlerFunc(f) is a Handler that calls f.

func (HandlerFunc) ProcessTask 

func (fn HandlerFunc) ProcessTask(ctx context.Context, task *Task) error

ProcessTask calls fn(ctx, task)

type Inspector 

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

Inspector is a client interface to inspect and mutate the state of queues and tasks.

func NewInspector 

func NewInspector(r RedisConnOpt) *Inspector

New returns a new instance of Inspector.

func NewInspectorFromRedisClient 

func NewInspectorFromRedisClient(c redis.UniversalClient) *Inspector

NewInspectorFromRedisClient returns a new instance of Inspector given a redis.UniversalClient Warning: The underlying redis connection pool will not be closed by Asynq, you are responsible for closing it.

func (*Inspector) ArchiveAllAggregatingTasks 

func (i *Inspector) ArchiveAllAggregatingTasks(queue, group string) (int, error)

ArchiveAllAggregatingTasks archives all tasks from the given group, and reports the number of tasks archived.

func (*Inspector) ArchiveAllPendingTasks 

func (i *Inspector) ArchiveAllPendingTasks(queue string) (int, error)

ArchiveAllPendingTasks archives all pending tasks from the given queue, and reports the number of tasks archived.

func (*Inspector) ArchiveAllRetryTasks 

func (i *Inspector) ArchiveAllRetryTasks(queue string) (int, error)

ArchiveAllRetryTasks archives all retry tasks from the given queue, and reports the number of tasks archiveed.

func (*Inspector) ArchiveAllScheduledTasks 

func (i *Inspector) ArchiveAllScheduledTasks(queue string) (int, error)

ArchiveAllScheduledTasks archives all scheduled tasks from the given queue, and reports the number of tasks archiveed.

func (*Inspector) ArchiveTask 

func (i *Inspector) ArchiveTask(queue, id string) error

ArchiveTask archives a task with the given id in the given queue. The task needs to be in pending, scheduled, or retry state, otherwise ArchiveTask will return an error.

If a queue with the given name doesn't exist, it returns an error wrapping ErrQueueNotFound. If a task with the given id doesn't exist in the queue, it returns an error wrapping ErrTaskNotFound. If the task is in already archived, it returns a non-nil error.

func (*Inspector) CancelProcessing 

func (i *Inspector) CancelProcessing(id string) error

CancelProcessing sends a signal to cancel processing of the task given a task id. CancelProcessing is best-effort, which means that it does not guarantee that the task with the given id will be canceled. The return value only indicates whether the cancelation signal has been sent.

func (*Inspector) Close 

func (i *Inspector) Close() error

Close closes the connection with redis.

func (*Inspector) ClusterKeySlot 

func (i *Inspector) ClusterKeySlot(queue string) (int64, error)

ClusterKeySlot returns an integer identifying the hash slot the given queue hashes to.

func (*Inspector) ClusterNodes 

func (i *Inspector) ClusterNodes(queue string) ([]*ClusterNode, error)

ClusterNodes returns a list of nodes the given queue belongs to.

Only relevant if task queues are stored in redis cluster.

func (*Inspector) DeleteAllAggregatingTasks 

func (i *Inspector) DeleteAllAggregatingTasks(queue, group string) (int, error)

DeleteAllAggregatingTasks deletes all tasks from the specified group, and reports the number of tasks deleted.

func (*Inspector) DeleteAllArchivedTasks 

func (i *Inspector) DeleteAllArchivedTasks(queue string) (int, error)

DeleteAllArchivedTasks deletes all archived tasks from the specified queue, and reports the number tasks deleted.

func (*Inspector) DeleteAllCompletedTasks 

func (i *Inspector) DeleteAllCompletedTasks(queue string) (int, error)

DeleteAllCompletedTasks deletes all completed tasks from the specified queue, and reports the number tasks deleted.

func (*Inspector) DeleteAllPendingTasks 

func (i *Inspector) DeleteAllPendingTasks(queue string) (int, error)

DeleteAllPendingTasks deletes all pending tasks from the specified queue, and reports the number tasks deleted.

func (*Inspector) DeleteAllRetryTasks 

func (i *Inspector) DeleteAllRetryTasks(queue string) (int, error)

DeleteAllRetryTasks deletes all retry tasks from the specified queue, and reports the number tasks deleted.

func (*Inspector) DeleteAllScheduledTasks 

func (i *Inspector) DeleteAllScheduledTasks(queue string) (int, error)

DeleteAllScheduledTasks deletes all scheduled tasks from the specified queue, and reports the number tasks deleted.

func (*Inspector) DeleteQueue 

func (i *Inspector) DeleteQueue(queue string, force bool) error

DeleteQueue removes the specified queue.

If force is set to true, DeleteQueue will remove the queue regardless of the queue size as long as no tasks are active in the queue. If force is set to false, DeleteQueue will remove the queue only if the queue is empty.

If the specified queue does not exist, DeleteQueue returns ErrQueueNotFound. If force is set to false and the specified queue is not empty, DeleteQueue returns ErrQueueNotEmpty.

func (*Inspector) DeleteTask 

func (i *Inspector) DeleteTask(queue, id string) error

DeleteTask deletes a task with the given id from the given queue. The task needs to be in pending, scheduled, retry, or archived state, otherwise DeleteTask will return an error.

If a queue with the given name doesn't exist, it returns an error wrapping ErrQueueNotFound. If a task with the given id doesn't exist in the queue, it returns an error wrapping ErrTaskNotFound. If the task is in active state, it returns a non-nil error.

func (*Inspector) GetQueueInfo 

func (i *Inspector) GetQueueInfo(queue string) (*QueueInfo, error)

GetQueueInfo returns current information of the given queue.

func (*Inspector) GetTaskInfo 

func (i *Inspector) GetTaskInfo(queue, id string) (*TaskInfo, error)

GetTaskInfo retrieves task information given a task id and queue name.

Returns an error wrapping ErrQueueNotFound if a queue with the given name doesn't exist. Returns an error wrapping ErrTaskNotFound if a task with the given id doesn't exist in the queue.

func (*Inspector) Groups 

func (i *Inspector) Groups(queue string) ([]*GroupInfo, error)

Groups returns a list of all groups within the given queue.

func (*Inspector) History 

func (i *Inspector) History(queue string, n int) ([]*DailyStats, error)

History returns a list of stats from the last n days.

func (*Inspector) ListActiveTasks 

func (i *Inspector) ListActiveTasks(queue string, opts ...ListOption) ([]*TaskInfo, error)

ListActiveTasks retrieves active tasks from the specified queue.

By default, it retrieves the first 30 tasks.

func (*Inspector) ListAggregatingTasks 

func (i *Inspector) ListAggregatingTasks(queue, group string, opts ...ListOption) ([]*TaskInfo, error)

ListAggregatingTasks retrieves scheduled tasks from the specified group.

By default, it retrieves the first 30 tasks.

func (*Inspector) ListArchivedTasks 

func (i *Inspector) ListArchivedTasks(queue string, opts ...ListOption) ([]*TaskInfo, error)

ListArchivedTasks retrieves archived tasks from the specified queue. Tasks are sorted by LastFailedAt in descending order.

By default, it retrieves the first 30 tasks.

func (*Inspector) ListCompletedTasks 

func (i *Inspector) ListCompletedTasks(queue string, opts ...ListOption) ([]*TaskInfo, error)

ListCompletedTasks retrieves completed tasks from the specified queue. Tasks are sorted by expiration time (i.e. CompletedAt + Retention) in descending order.

By default, it retrieves the first 30 tasks.

func (*Inspector) ListPendingTasks 

func (i *Inspector) ListPendingTasks(queue string, opts ...ListOption) ([]*TaskInfo, error)

ListPendingTasks retrieves pending tasks from the specified queue.

By default, it retrieves the first 30 tasks.

func (*Inspector) ListRetryTasks 

func (i *Inspector) ListRetryTasks(queue string, opts ...ListOption) ([]*TaskInfo, error)

ListRetryTasks retrieves retry tasks from the specified queue. Tasks are sorted by NextProcessAt in ascending order.

By default, it retrieves the first 30 tasks.

func (*Inspector) ListScheduledTasks 

func (i *Inspector) ListScheduledTasks(queue string, opts ...ListOption) ([]*TaskInfo, error)

ListScheduledTasks retrieves scheduled tasks from the specified queue. Tasks are sorted by NextProcessAt in ascending order.

By default, it retrieves the first 30 tasks.

func (*Inspector) ListSchedulerEnqueueEvents 

func (i *Inspector) ListSchedulerEnqueueEvents(entryID string, opts ...ListOption) ([]*SchedulerEnqueueEvent, error)

ListSchedulerEnqueueEvents retrieves a list of enqueue events from the specified scheduler entry.

By default, it retrieves the first 30 tasks.

func (*Inspector) PauseQueue 

func (i *Inspector) PauseQueue(queue string) error

PauseQueue pauses task processing on the specified queue. If the queue is already paused, it will return a non-nil error.

func (*Inspector) Queues 

func (i *Inspector) Queues() ([]string, error)

Queues returns a list of all queue names.

func (*Inspector) RunAllAggregatingTasks 

func (i *Inspector) RunAllAggregatingTasks(queue, group string) (int, error)

RunAllAggregatingTasks schedules all tasks from the given grou to run. and reports the number of tasks scheduled to run.

func (*Inspector) RunAllArchivedTasks 

func (i *Inspector) RunAllArchivedTasks(queue string) (int, error)

RunAllArchivedTasks schedules all archived tasks from the given queue to run, and reports the number of tasks scheduled to run.

func (*Inspector) RunAllRetryTasks 

func (i *Inspector) RunAllRetryTasks(queue string) (int, error)

RunAllRetryTasks schedules all retry tasks from the given queue to run, and reports the number of tasks scheduled to run.

func (*Inspector) RunAllScheduledTasks 

func (i *Inspector) RunAllScheduledTasks(queue string) (int, error)

RunAllScheduledTasks schedules all scheduled tasks from the given queue to run, and reports the number of tasks scheduled to run.

func (*Inspector) RunTask 

func (i *Inspector) RunTask(queue, id string) error

RunTask updates the task to pending state given a queue name and task id. The task needs to be in scheduled, retry, or archived state, otherwise RunTask will return an error.

If a queue with the given name doesn't exist, it returns an error wrapping ErrQueueNotFound. If a task with the given id doesn't exist in the queue, it returns an error wrapping ErrTaskNotFound. If the task is in pending or active state, it returns a non-nil error.

func (*Inspector) SchedulerEntries 

func (i *Inspector) SchedulerEntries() ([]*SchedulerEntry, error)

SchedulerEntries returns a list of all entries registered with currently running schedulers.

func (*Inspector) Servers 

func (i *Inspector) Servers() ([]*ServerInfo, error)

Servers return a list of running servers' information.

func (*Inspector) UnpauseQueue 

func (i *Inspector) UnpauseQueue(queue string) error

UnpauseQueue resumes task processing on the specified queue. If the queue is not paused, it will return a non-nil error.

type ListOption 

type ListOption interface{}

ListOption specifies behavior of list operation.

func Page 

func Page(n int) ListOption

Page returns an option to specify the page number for list operation. The value 1 fetches the first page.

Negative page number is treated as one.

func PageSize 

func PageSize(n int) ListOption

PageSize returns an option to specify the page size for list operation.

Negative page size is treated as zero.

type LogLevel 

type LogLevel int32

LogLevel represents logging level.

It satisfies flag.Value interface. 

const (

	// DebugLevel is the lowest level of logging.
	// Debug logs are intended for debugging and development purposes.
	DebugLevel LogLevel

	// InfoLevel is used for general informational log messages.
	InfoLevel

	// WarnLevel is used for undesired but relatively expected events,
	// which may indicate a problem.
	WarnLevel

	// ErrorLevel is used for undesired and unexpected events that
	// the program can recover from.
	ErrorLevel

	// FatalLevel is used for undesired and unexpected events that
	// the program cannot recover from.
	FatalLevel
)

func (*LogLevel) Set 

func (l *LogLevel) Set(val string) error

Set is part of the flag.Value interface.

func (*LogLevel) String 

func (l *LogLevel) String() string

String is part of the flag.Value interface.

type Logger 

type Logger interface {
	// Debug logs a message at Debug level.
	Debug(args ...interface{})

	// Info logs a message at Info level.
	Info(args ...interface{})

	// Warn logs a message at Warning level.
	Warn(args ...interface{})

	// Error logs a message at Error level.
	Error(args ...interface{})

	// Fatal logs a message at Fatal level
	// and process will exit with status set to 1.
	Fatal(args ...interface{})
}

Logger supports logging at various log levels.

type MiddlewareFunc 

type MiddlewareFunc func(Handler) Handler

MiddlewareFunc is a function which receives an asynq.Handler and returns another asynq.Handler. Typically, the returned handler is a closure which does something with the context and task passed to it, and then calls the handler passed as parameter to the MiddlewareFunc.

type Option 

type Option interface {
	// String returns a string representation of the option.
	String() string

	// Type describes the type of the option.
	Type() OptionType

	// Value returns a value used to create this option.
	Value() interface{}
}

Option specifies the task processing behavior.

func Deadline 

func Deadline(t time.Time) Option

Deadline returns an option to specify the deadline for the given task. If it reaches the deadline before the Handler returns, then the task will be retried.

If there's a conflicting Timeout option, whichever comes earliest will be used.

func Group 

func Group(name string) Option

Group returns an option to specify the group used for the task. Tasks in a given queue with the same group will be aggregated into one task before passed to Handler.

func MaxRetry 

func MaxRetry(n int) Option

MaxRetry returns an option to specify the max number of times the task will be retried.

Negative retry count is treated as zero retry.

func ProcessAt 

func ProcessAt(t time.Time) Option

ProcessAt returns an option to specify when to process the given task.

If there's a conflicting ProcessIn option, the last option passed to Enqueue overrides the others.

func ProcessIn 

func ProcessIn(d time.Duration) Option

ProcessIn returns an option to specify when to process the given task relative to the current time.

If there's a conflicting ProcessAt option, the last option passed to Enqueue overrides the others.

func Queue 

func Queue(name string) Option

Queue returns an option to specify the queue to enqueue the task into.

func Retention 

func Retention(d time.Duration) Option

Retention returns an option to specify the duration of retention period for the task. If this option is provided, the task will be stored as a completed task after successful processing. A completed task will be deleted after the specified duration elapses.

func TaskID 

func TaskID(id string) Option

TaskID returns an option to specify the task ID.

func Timeout 

func Timeout(d time.Duration) Option

Timeout returns an option to specify how long a task may run. If the timeout elapses before the Handler returns, then the task will be retried.

Zero duration means no limit.

If there's a conflicting Deadline option, whichever comes earliest will be used.

func Unique 

func Unique(ttl time.Duration) Option

Unique returns an option to enqueue a task only if the given task is unique. Task enqueued with this option is guaranteed to be unique within the given ttl. Once the task gets processed successfully or once the TTL has expired, another task with the same uniqueness may be enqueued. ErrDuplicateTask error is returned when enqueueing a duplicate task. TTL duration must be greater than or equal to 1 second.

Uniqueness of a task is based on the following properties:

  • Task Type
  • Task Payload
  • Queue Name

type OptionType 

type OptionType int
const (
	MaxRetryOpt OptionType = iota
	QueueOpt
	TimeoutOpt
	DeadlineOpt
	UniqueOpt
	ProcessAtOpt
	ProcessInOpt
	TaskIDOpt
	RetentionOpt
	GroupOpt
)

type PeriodicTaskConfig 

type PeriodicTaskConfig struct {
	Cronspec string   // required: must be non empty string
	Task     *Task    // required: must be non nil
	Opts     []Option // optional: can be nil
}

PeriodicTaskConfig specifies the details of a periodic task.

type PeriodicTaskConfigProvider 

type PeriodicTaskConfigProvider interface {
	GetConfigs() ([]*PeriodicTaskConfig, error)
}

PeriodicTaskConfigProvider provides configs for periodic tasks. GetConfigs will be called by a PeriodicTaskManager periodically to sync the scheduler's entries with the configs returned by the provider.

type PeriodicTaskManager 

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

PeriodicTaskManager manages scheduling of periodic tasks. It syncs scheduler's entries by calling the config provider periodically.

func NewPeriodicTaskManager 

func NewPeriodicTaskManager(opts PeriodicTaskManagerOpts) (*PeriodicTaskManager, error)

NewPeriodicTaskManager returns a new PeriodicTaskManager instance. The given opts should specify the RedisConnOp and PeriodicTaskConfigProvider at minimum.

func (*PeriodicTaskManager) Run 

func (mgr *PeriodicTaskManager) Run() error

Run starts the manager and blocks until an os signal to exit the program is received. Once it receives a signal, it gracefully shuts down the manager.

func (*PeriodicTaskManager) Shutdown 

func (mgr *PeriodicTaskManager) Shutdown()

Shutdown gracefully shuts down the manager. It notifies a background syncer goroutine to stop and stops scheduler.

func (*PeriodicTaskManager) Start 

func (mgr *PeriodicTaskManager) Start() error

Start starts a scheduler and background goroutine to sync the scheduler with the configs returned by the provider.

Start returns any error encountered at start up time.

type PeriodicTaskManagerOpts 

type PeriodicTaskManagerOpts struct {
	// Required: must be non nil
	PeriodicTaskConfigProvider PeriodicTaskConfigProvider

	// Optional: if RedisUniversalClient is nil must be non nil
	RedisConnOpt RedisConnOpt

	// Optional: if RedisUniversalClient is non nil, RedisConnOpt is ignored.
	RedisUniversalClient redis.UniversalClient

	// Optional: scheduler options
	*SchedulerOpts

	// Optional: default is 3m
	SyncInterval time.Duration
}

type QueueInfo 

type QueueInfo struct {
	// Name of the queue.
	Queue string

	// Total number of bytes that the queue and its tasks require to be stored in redis.
	// It is an approximate memory usage value in bytes since the value is computed by sampling.
	MemoryUsage int64

	// Latency of the queue, measured by the oldest pending task in the queue.
	Latency time.Duration

	// Size is the total number of tasks in the queue.
	// The value is the sum of Pending, Active, Scheduled, Retry, Aggregating and Archived.
	Size int

	// Groups is the total number of groups in the queue.
	Groups int

	// Number of pending tasks.
	Pending int
	// Number of active tasks.
	Active int
	// Number of scheduled tasks.
	Scheduled int
	// Number of retry tasks.
	Retry int
	// Number of archived tasks.
	Archived int
	// Number of stored completed tasks.
	Completed int
	// Number of aggregating tasks.
	Aggregating int

	// Total number of tasks being processed within the given date (counter resets daily).
	// The number includes both succeeded and failed tasks.
	Processed int
	// Total number of tasks failed to be processed within the given date (counter resets daily).
	Failed int

	// Total number of tasks processed (cumulative).
	ProcessedTotal int
	// Total number of tasks failed (cumulative).
	FailedTotal int

	// Paused indicates whether the queue is paused.
	// If true, tasks in the queue will not be processed.
	Paused bool

	// Time when this queue info snapshot was taken.
	Timestamp time.Time
}

QueueInfo represents a state of a queue at a certain time.

type RedisClientOpt 

type RedisClientOpt struct {
	// Network type to use, either tcp or unix.
	// Default is tcp.
	Network string

	// Redis server address in "host:port" format.
	Addr string

	// Username to authenticate the current connection when Redis ACLs are used.
	// See: https://redis.io/commands/auth.
	Username string

	// Password to authenticate the current connection.
	// See: https://redis.io/commands/auth.
	Password string

	// Redis DB to select after connecting to a server.
	// See: https://redis.io/commands/select.
	DB int

	// Dial timeout for establishing new connections.
	// Default is 5 seconds.
	DialTimeout time.Duration

	// Timeout for socket reads.
	// If timeout is reached, read commands will fail with a timeout error
	// instead of blocking.
	//
	// Use value -1 for no timeout and 0 for default.
	// Default is 3 seconds.
	ReadTimeout time.Duration

	// Timeout for socket writes.
	// If timeout is reached, write commands will fail with a timeout error
	// instead of blocking.
	//
	// Use value -1 for no timeout and 0 for default.
	// Default is ReadTimout.
	WriteTimeout time.Duration

	// Maximum number of socket connections.
	// Default is 10 connections per every CPU as reported by runtime.NumCPU.
	PoolSize int

	// TLS Config used to connect to a server.
	// TLS will be negotiated only if this field is set.
	TLSConfig *tls.Config
}

RedisClientOpt is used to create a redis client that connects to a redis server directly.

func (RedisClientOpt) MakeRedisClient 

func (opt RedisClientOpt) MakeRedisClient() interface{}

type RedisClusterClientOpt 

type RedisClusterClientOpt struct {
	// A seed list of host:port addresses of cluster nodes.
	Addrs []string

	// The maximum number of retries before giving up.
	// Command is retried on network errors and MOVED/ASK redirects.
	// Default is 8 retries.
	MaxRedirects int

	// Username to authenticate the current connection when Redis ACLs are used.
	// See: https://redis.io/commands/auth.
	Username string

	// Password to authenticate the current connection.
	// See: https://redis.io/commands/auth.
	Password string

	// Dial timeout for establishing new connections.
	// Default is 5 seconds.
	DialTimeout time.Duration

	// Timeout for socket reads.
	// If timeout is reached, read commands will fail with a timeout error
	// instead of blocking.
	//
	// Use value -1 for no timeout and 0 for default.
	// Default is 3 seconds.
	ReadTimeout time.Duration

	// Timeout for socket writes.
	// If timeout is reached, write commands will fail with a timeout error
	// instead of blocking.
	//
	// Use value -1 for no timeout and 0 for default.
	// Default is ReadTimeout.
	WriteTimeout time.Duration

	// TLS Config used to connect to a server.
	// TLS will be negotiated only if this field is set.
	TLSConfig *tls.Config
}

RedisClusterClientOpt is used to creates a redis client that connects to redis cluster.

func (RedisClusterClientOpt) MakeRedisClient 

func (opt RedisClusterClientOpt) MakeRedisClient() interface{}

type RedisConnOpt 

type RedisConnOpt interface {
	// MakeRedisClient returns a new redis client instance.
	// Return value is intentionally opaque to hide the implementation detail of redis client.
	MakeRedisClient() interface{}
}

RedisConnOpt is a discriminated union of types that represent Redis connection configuration option.

RedisConnOpt represents a sum of following types:

  • RedisClientOpt
  • RedisFailoverClientOpt
  • RedisClusterClientOpt

func ParseRedisURI 

func ParseRedisURI(uri string) (RedisConnOpt, error)

ParseRedisURI parses redis uri string and returns RedisConnOpt if uri is valid. It returns a non-nil error if uri cannot be parsed.

Three URI schemes are supported, which are redis:, rediss:, redis-socket:, and redis-sentinel:. Supported formats are: 

redis://[:password@]host[:port][/dbnumber]
rediss://[:password@]host[:port][/dbnumber]
redis-socket://[:password@]path[?db=dbnumber]
redis-sentinel://[:password@]host1[:port][,host2:[:port]][,hostN:[:port]][?master=masterName]
Example 
package main

import (
	"fmt"
	"log"

	"github.com/OlegBugaichuk/asynq"
)

func main() {
	rconn, err := asynq.ParseRedisURI("redis://localhost:6379/10")
	if err != nil {
		log.Fatal(err)
	}
	r, ok := rconn.(asynq.RedisClientOpt)
	if !ok {
		log.Fatal("unexpected type")
	}
	fmt.Println(r.Addr)
	fmt.Println(r.DB)
}

Output:

localhost:6379
10

type RedisFailoverClientOpt 

type RedisFailoverClientOpt struct {
	// Redis master name that monitored by sentinels.
	MasterName string

	// Addresses of sentinels in "host:port" format.
	// Use at least three sentinels to avoid problems described in
	// https://redis.io/topics/sentinel.
	SentinelAddrs []string

	// Redis sentinel username.
	SentinelUsername string

	// Redis sentinel password.
	SentinelPassword string

	// Username to authenticate the current connection when Redis ACLs are used.
	// See: https://redis.io/commands/auth.
	Username string

	// Password to authenticate the current connection.
	// See: https://redis.io/commands/auth.
	Password string

	// Redis DB to select after connecting to a server.
	// See: https://redis.io/commands/select.
	DB int

	// Dial timeout for establishing new connections.
	// Default is 5 seconds.
	DialTimeout time.Duration

	// Timeout for socket reads.
	// If timeout is reached, read commands will fail with a timeout error
	// instead of blocking.
	//
	// Use value -1 for no timeout and 0 for default.
	// Default is 3 seconds.
	ReadTimeout time.Duration

	// Timeout for socket writes.
	// If timeout is reached, write commands will fail with a timeout error
	// instead of blocking.
	//
	// Use value -1 for no timeout and 0 for default.
	// Default is ReadTimeout
	WriteTimeout time.Duration

	// Maximum number of socket connections.
	// Default is 10 connections per every CPU as reported by runtime.NumCPU.
	PoolSize int

	// TLS Config used to connect to a server.
	// TLS will be negotiated only if this field is set.
	TLSConfig *tls.Config
}

RedisFailoverClientOpt is used to creates a redis client that talks to redis sentinels for service discovery and has an automatic failover capability.

func (RedisFailoverClientOpt) MakeRedisClient 

func (opt RedisFailoverClientOpt) MakeRedisClient() interface{}

type ResultWriter 

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

ResultWriter is a client interface to write result data for a task. It writes the data to the redis instance the server is connected to.

Example 
package main

import (
	"context"
	"fmt"
	"log"

	"github.com/OlegBugaichuk/asynq"
)

func main() {
	// ResultWriter is only accessible in Handler.
	h := func(ctx context.Context, task *asynq.Task) error {
		// .. do task processing work

		res := []byte("task result data")
		n, err := task.ResultWriter().Write(res) // implements io.Writer
		if err != nil {
			return fmt.Errorf("failed to write task result: %v", err)
		}
		log.Printf(" %d bytes written", n)
		return nil
	}

	_ = h
}

Output:

func (*ResultWriter) TaskID 

func (w *ResultWriter) TaskID() string

TaskID returns the ID of the task the ResultWriter is associated with.

func (*ResultWriter) Write 

func (w *ResultWriter) Write(data []byte) (n int, err error)

Write writes the given data as a result of the task the ResultWriter is associated with.

type RetryDelayFunc 

type RetryDelayFunc func(n int, e error, t *Task) time.Duration

RetryDelayFunc calculates the retry delay duration for a failed task given the retry count, error, and the task.

n is the number of times the task has been retried. e is the error returned by the task handler. t is the task in question.

type Scheduler 

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

A Scheduler kicks off tasks at regular intervals based on the user defined schedule.

Schedulers are safe for concurrent use by multiple goroutines.

Example 
package main

import (
	"log"
	"time"

	"github.com/OlegBugaichuk/asynq"
)

func main() {
	scheduler := asynq.NewScheduler(
		asynq.RedisClientOpt{Addr: ":6379"},
		&asynq.SchedulerOpts{Location: time.Local},
	)

	if _, err := scheduler.Register("* * * * *", asynq.NewTask("task1", nil)); err != nil {
		log.Fatal(err)
	}
	if _, err := scheduler.Register("@every 30s", asynq.NewTask("task2", nil)); err != nil {
		log.Fatal(err)
	}

	// Run blocks and waits for os signal to terminate the program.
	if err := scheduler.Run(); err != nil {
		log.Fatal(err)
	}
}

Output:

func NewScheduler 

func NewScheduler(r RedisConnOpt, opts *SchedulerOpts) *Scheduler

NewScheduler returns a new Scheduler instance given the redis connection option. The parameter opts is optional, defaults will be used if opts is set to nil

func NewSchedulerFromRedisClient 

func NewSchedulerFromRedisClient(c redis.UniversalClient, opts *SchedulerOpts) *Scheduler

NewSchedulerFromRedisClient returns a new instance of Scheduler given a redis.UniversalClient The parameter opts is optional, defaults will be used if opts is set to nil. Warning: The underlying redis connection pool will not be closed by Asynq, you are responsible for closing it.

func (*Scheduler) Ping 

func (s *Scheduler) Ping() error

Ping performs a ping against the redis connection.

func (*Scheduler) Register 

func (s *Scheduler) Register(cronspec string, task *Task, opts ...Option) (entryID string, err error)

Register registers a task to be enqueued on the given schedule specified by the cronspec. It returns an ID of the newly registered entry.

func (*Scheduler) Run 

func (s *Scheduler) Run() error

Run starts the scheduler until an os signal to exit the program is received. It returns an error if scheduler is already running or has been shutdown.

func (*Scheduler) Shutdown 

func (s *Scheduler) Shutdown()

Shutdown stops and shuts down the scheduler.

func (*Scheduler) Start 

func (s *Scheduler) Start() error

Start starts the scheduler. It returns an error if the scheduler is already running or has been shutdown.

func (*Scheduler) Unregister 

func (s *Scheduler) Unregister(entryID string) error

Unregister removes a registered entry by entry ID. Unregister returns a non-nil error if no entries were found for the given entryID.

type SchedulerEnqueueEvent 

type SchedulerEnqueueEvent struct {
	// ID of the task that was enqueued.
	TaskID string

	// Time the task was enqueued.
	EnqueuedAt time.Time
}

SchedulerEnqueueEvent holds information about an enqueue event by a scheduler.

type SchedulerEntry 

type SchedulerEntry struct {
	// Identifier of this entry.
	ID string

	// Spec describes the schedule of this entry.
	Spec string

	// Periodic Task registered for this entry.
	Task *Task

	// Opts is the options for the periodic task.
	Opts []Option

	// Next shows the next time the task will be enqueued.
	Next time.Time

	// Prev shows the last time the task was enqueued.
	// Zero time if task was never enqueued.
	Prev time.Time
}

SchedulerEntry holds information about a periodic task registered with a scheduler.

type SchedulerOpts 

type SchedulerOpts struct {
	// HeartbeatInterval specifies the interval between scheduler heartbeats.
	//
	// If unset, zero or a negative value, the interval is set to 10 second.
	//
	// Note: Setting this value too low may add significant load to redis.
	//
	// By default, HeartbeatInterval is set to 10 seconds.
	HeartbeatInterval time.Duration

	// Logger specifies the logger used by the scheduler instance.
	//
	// If unset, the default logger is used.
	Logger Logger

	// LogLevel specifies the minimum log level to enable.
	//
	// If unset, InfoLevel is used by default.
	LogLevel LogLevel

	// Location specifies the time zone location.
	//
	// If unset, the UTC time zone (time.UTC) is used.
	Location *time.Location

	// PreEnqueueFunc, if provided, is called before a task gets enqueued by Scheduler.
	// The callback function should return quickly to not block the current thread.
	PreEnqueueFunc func(task *Task, opts []Option)

	// PostEnqueueFunc, if provided, is called after a task gets enqueued by Scheduler.
	// The callback function should return quickly to not block the current thread.
	PostEnqueueFunc func(info *TaskInfo, err error)

	// Deprecated: Use PostEnqueueFunc instead
	// EnqueueErrorHandler gets called when scheduler cannot enqueue a registered task
	// due to an error.
	EnqueueErrorHandler func(task *Task, opts []Option, err error)
}

SchedulerOpts specifies scheduler options.

type ServeMux 

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

ServeMux is a multiplexer for asynchronous tasks. It matches the type of each task against a list of registered patterns and calls the handler for the pattern that most closely matches the task's type name.

Longer patterns take precedence over shorter ones, so that if there are handlers registered for both "images" and "images:thumbnails", the latter handler will be called for tasks with a type name beginning with "images:thumbnails" and the former will receive tasks with type name beginning with "images".

func NewServeMux 

func NewServeMux() *ServeMux

NewServeMux allocates and returns a new ServeMux.

func (*ServeMux) Handle 

func (mux *ServeMux) Handle(pattern string, handler Handler)

Handle registers the handler for the given pattern. If a handler already exists for pattern, Handle panics.

func (*ServeMux) HandleFunc 

func (mux *ServeMux) HandleFunc(pattern string, handler func(context.Context, *Task) error)

HandleFunc registers the handler function for the given pattern.

func (*ServeMux) Handler 

func (mux *ServeMux) Handler(t *Task) (h Handler, pattern string)

Handler returns the handler to use for the given task. It always return a non-nil handler.

Handler also returns the registered pattern that matches the task.

If there is no registered handler that applies to the task, handler returns a 'not found' handler which returns an error.

func (*ServeMux) ProcessTask 

func (mux *ServeMux) ProcessTask(ctx context.Context, task *Task) error

ProcessTask dispatches the task to the handler whose pattern most closely matches the task type.

func (*ServeMux) Use 

func (mux *ServeMux) Use(mws ...MiddlewareFunc)

Use appends a MiddlewareFunc to the chain. Middlewares are executed in the order that they are applied to the ServeMux.

type Server 

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

Server is responsible for task processing and task lifecycle management.

Server pulls tasks off queues and processes them. If the processing of a task is unsuccessful, server will schedule it for a retry.

A task will be retried until either the task gets processed successfully or until it reaches its max retry count.

If a task exhausts its retries, it will be moved to the archive and will be kept in the archive set. Note that the archive size is finite and once it reaches its max size, oldest tasks in the archive will be deleted.

func NewServer 

func NewServer(r RedisConnOpt, cfg Config) *Server

NewServer returns a new Server given a redis connection option and server configuration.

func NewServerFromRedisClient 

func NewServerFromRedisClient(c redis.UniversalClient, cfg Config) *Server

NewServerFromRedisClient returns a new instance of Server given a redis.UniversalClient and server configuration Warning: The underlying redis connection pool will not be closed by Asynq, you are responsible for closing it.

func (*Server) Ping 

func (srv *Server) Ping() error

Ping performs a ping against the redis connection.

This is an alternative to the HealthCheckFunc available in the Config object.

func (*Server) Run 

func (srv *Server) Run(handler Handler) error

Run starts the task processing and blocks until an os signal to exit the program is received. Once it receives a signal, it gracefully shuts down all active workers and other goroutines to process the tasks.

Run returns any error encountered at server startup time. If the server has already been shutdown, ErrServerClosed is returned.

Example 
package main

import (
	"log"

	"github.com/OlegBugaichuk/asynq"
)

func main() {
	srv := asynq.NewServer(
		asynq.RedisClientOpt{Addr: ":6379"},
		asynq.Config{Concurrency: 20},
	)

	h := asynq.NewServeMux()
	// ... Register handlers

	// Run blocks and waits for os signal to terminate the program.
	if err := srv.Run(h); err != nil {
		log.Fatal(err)
	}
}

Output:

func (*Server) Shutdown 

func (srv *Server) Shutdown()

Shutdown gracefully shuts down the server. It gracefully closes all active workers. The server will wait for active workers to finish processing tasks for duration specified in Config.ShutdownTimeout. If worker didn't finish processing a task during the timeout, the task will be pushed back to Redis.

Example 
package main

import (
	"log"
	"os"
	"os/signal"

	"github.com/OlegBugaichuk/asynq"
	"golang.org/x/sys/unix"
)

func main() {
	srv := asynq.NewServer(
		asynq.RedisClientOpt{Addr: ":6379"},
		asynq.Config{Concurrency: 20},
	)

	h := asynq.NewServeMux()
	// ... Register handlers

	if err := srv.Start(h); err != nil {
		log.Fatal(err)
	}

	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, unix.SIGTERM, unix.SIGINT)
	<-sigs // wait for termination signal

	srv.Shutdown()
}

Output:

func (*Server) Start 

func (srv *Server) Start(handler Handler) error

Start starts the worker server. Once the server has started, it pulls tasks off queues and starts a worker goroutine for each task and then call Handler to process it. Tasks are processed concurrently by the workers up to the number of concurrency specified in Config.Concurrency.

Start returns any error encountered at server startup time. If the server has already been shutdown, ErrServerClosed is returned.

func (*Server) Stop 

func (srv *Server) Stop()

Stop signals the server to stop pulling new tasks off queues. Stop can be used before shutting down the server to ensure that all currently active tasks are processed before server shutdown.

Stop does not shutdown the server, make sure to call Shutdown before exit.

Example 
package main

import (
	"log"
	"os"
	"os/signal"

	"github.com/OlegBugaichuk/asynq"
	"golang.org/x/sys/unix"
)

func main() {
	srv := asynq.NewServer(
		asynq.RedisClientOpt{Addr: ":6379"},
		asynq.Config{Concurrency: 20},
	)

	h := asynq.NewServeMux()
	// ... Register handlers

	if err := srv.Start(h); err != nil {
		log.Fatal(err)
	}

	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, unix.SIGTERM, unix.SIGINT, unix.SIGTSTP)
	// Handle SIGTERM, SIGINT to exit the program.
	// Handle SIGTSTP to stop processing new tasks.
	for {
		s := <-sigs
		if s == unix.SIGTSTP {
			srv.Stop() // stop processing new tasks
			continue
		}
		break // received SIGTERM or SIGINT signal
	}

	srv.Shutdown()
}

Output:

type ServerInfo 

type ServerInfo struct {
	// Unique Identifier for the server.
	ID string
	// Host machine on which the server is running.
	Host string
	// PID of the process in which the server is running.
	PID int

	// Server configuration details.
	// See Config doc for field descriptions.
	Concurrency    int
	Queues         map[string]int
	StrictPriority bool

	// Time the server started.
	Started time.Time
	// Status indicates the status of the server.
	// TODO: Update comment with more details.
	Status string
	// A List of active workers currently processing tasks.
	ActiveWorkers []*WorkerInfo
}

ServerInfo describes a running Server instance.

type Task 

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

Task represents a unit of work to be performed.

func NewTask 

func NewTask(typename string, payload []byte, opts ...Option) *Task

NewTask returns a new Task given a type name and payload data. Options can be passed to configure task processing behavior.

func (*Task) Payload 

func (t *Task) Payload() []byte

func (*Task) ResultWriter 

func (t *Task) ResultWriter() *ResultWriter

ResultWriter returns a pointer to the ResultWriter associated with the task.

Nil pointer is returned if called on a newly created task (i.e. task created by calling NewTask). Only the tasks passed to Handler.ProcessTask have a valid ResultWriter pointer.

func (*Task) Type 

func (t *Task) Type() string

type TaskInfo 

type TaskInfo struct {
	// ID is the identifier of the task.
	ID string

	// Queue is the name of the queue in which the task belongs.
	Queue string

	// Type is the type name of the task.
	Type string

	// Payload is the payload data of the task.
	Payload []byte

	// State indicates the task state.
	State TaskState

	// MaxRetry is the maximum number of times the task can be retried.
	MaxRetry int

	// Retried is the number of times the task has retried so far.
	Retried int

	// LastErr is the error message from the last failure.
	LastErr string

	// LastFailedAt is the time time of the last failure if any.
	// If the task has no failures, LastFailedAt is zero time (i.e. time.Time{}).
	LastFailedAt time.Time

	// Timeout is the duration the task can be processed by Handler before being retried,
	// zero if not specified
	Timeout time.Duration

	// Deadline is the deadline for the task, zero value if not specified.
	Deadline time.Time

	// Group is the name of the group in which the task belongs.
	//
	// Tasks in the same queue can be grouped together by Group name and will be aggregated into one task
	// by a Server processing the queue.
	//
	// Empty string (default) indicates task does not belong to any groups, and no aggregation will be applied to the task.
	Group string

	// NextProcessAt is the time the task is scheduled to be processed,
	// zero if not applicable.
	NextProcessAt time.Time

	// IsOrphaned describes whether the task is left in active state with no worker processing it.
	// An orphaned task indicates that the worker has crashed or experienced network failures and was not able to
	// extend its lease on the task.
	//
	// This task will be recovered by running a server against the queue the task is in.
	// This field is only applicable to tasks with TaskStateActive.
	IsOrphaned bool

	// Retention is duration of the retention period after the task is successfully processed.
	Retention time.Duration

	// CompletedAt is the time when the task is processed successfully.
	// Zero value (i.e. time.Time{}) indicates no value.
	CompletedAt time.Time

	// Result holds the result data associated with the task.
	// Use ResultWriter to write result data from the Handler.
	Result []byte
}

A TaskInfo describes a task and its metadata.

type TaskState 

type TaskState int

TaskState denotes the state of a task. 

const (
	// Indicates that the task is currently being processed by Handler.
	TaskStateActive TaskState = iota + 1

	// Indicates that the task is ready to be processed by Handler.
	TaskStatePending

	// Indicates that the task is scheduled to be processed some time in the future.
	TaskStateScheduled

	// Indicates that the task has previously failed and scheduled to be processed some time in the future.
	TaskStateRetry

	// Indicates that the task is archived and stored for inspection purposes.
	TaskStateArchived

	// Indicates that the task is processed successfully and retained until the retention TTL expires.
	TaskStateCompleted

	// Indicates that the task is waiting in a group to be aggregated into one task.
	TaskStateAggregating
)

func (TaskState) String 

func (s TaskState) String() string

type WorkerInfo 

type WorkerInfo struct {
	// ID of the task the worker is processing.
	TaskID string
	// Type of the task the worker is processing.
	TaskType string
	// Payload of the task the worker is processing.
	TaskPayload []byte
	// Queue from which the worker got its task.
	Queue string
	// Time the worker started processing the task.
	Started time.Time
	// Time the worker needs to finish processing the task by.
	Deadline time.Time
}

WorkerInfo describes a running worker processing a task.

Source Files

View all Source files

Directories

Path Synopsis
internal
base
Package base defines foundational types and constants used in asynq package.
 Package base defines foundational types and constants used in asynq package.
context
errors
Package errors defines the error type and functions used by asynq and its internal packages.
 Package errors defines the error type and functions used by asynq and its internal packages.
log
Package log exports logging related types and functions.
 Package log exports logging related types and functions.
proto
rdb
Package rdb encapsulates the interactions with redis.
 Package rdb encapsulates the interactions with redis.
testbroker
Package testbroker exports a broker implementation that should be used in package testing.
 Package testbroker exports a broker implementation that should be used in package testing.
testutil
Package testutil defines test helpers for asynq and its internal packages.
 Package testutil defines test helpers for asynq and its internal packages.
timeutil
Package timeutil exports functions and types related to time and date.
 Package timeutil exports functions and types related to time and date.
x module

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.