Back to godoc.org

Package tq

v0.0.0-...-1557bcf
Latest Go to latest

The latest major version is .

Published: Aug 13, 2020 | License: Apache-2.0 | Module: github.com/luci/luci-go

Overview

Package tq provides a task queue implementation on top of Cloud Tasks.

It exposes a high-level API that operates with proto messages and hides gory details such as serialization, routing, authentication, etc.

Enabling it on Appengine

All default options should just work, except you still need to setup the sweeper cron and the sweeper queue.

In cron.yaml:

- url: /internal/tasks/c/sweep
  schedule: every 1 minutes

In queue.yaml:

- name: tq-sweep
  rate: 500/s

Using it with Spanner

You need to Create a table in your database:

CREATE TABLE TQReminders (
  ID STRING(MAX) NOT NULL,
  FreshUntil TIMESTAMP NOT NULL,
  Payload BYTES(102400) NOT NULL,
  Extra BYTES(1024) NOT NULL,
) PRIMARY KEY (ID ASC);

Index

Variables

var Retry = errors.BoolTag{Key: errors.NewTagKey("the task should be retried")}

Retry is an error tag used to indicate that the handler wants the task to be redelivered later.

See Handler doc for more details.

func AddTask

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

AddTask is a shortcut for Default.AddTask.

func MustAddTask

func MustAddTask(ctx context.Context, task *Task)

MustAddTask is like AddTask, but panics on errors.

Mostly useful for AddTask calls inside a Spanner transaction, where they essentially just call span.BufferWrite (i.e. make no RPCs) and thus can fail only if arguments are bad (in which case it is OK to panic).

func NewModule

func NewModule(opts *ModuleOptions) module.Module

NewModule returns a server module that sets up a TQ dispatcher.

func NewModuleFromFlags

func NewModuleFromFlags() module.Module

NewModuleFromFlags is a variant of NewModule that initializes options through command line flags.

Calling this function registers flags in flag.CommandLine. They are usually parsed in server.Main(...).

func Sweep

func Sweep(ctx context.Context) error

Sweep is a shortcut for Default.Sweep.

func TestingContext

func TestingContext(ctx context.Context, d *Dispatcher) (context.Context, *tqtesting.Scheduler)

TestingContext creates a scheduler that executes tasks through the given dispatcher (or Default one if nil) and puts it into the context as Submitter, so AddTask calls eventually submit tasks into this scheduler.

The end result is that tasks submitted using such context end up in the returned Scheduler (allowing them to be examined), and when the Scheduler delivers them, they result in calls to corresponding handlers registered in the Dispatcher.

func UseSubmitter

func UseSubmitter(ctx context.Context, s Submitter) context.Context

UseSubmitter puts an arbitrary submitter in the context.

It will be used by Dispatcher's AddTask to submit Cloud Tasks.

func ValidateNamespace

func ValidateNamespace(n string) error

ValidateNamespace returns an error if `n` is not a valid namespace name.

An empty string is a valid namespace (denoting the default namespace). Other valid namespaces must start with an ASCII letter or '_', contain only ASCII letters, digits or '_', and be less than 50 chars in length.

type CloudTaskSubmitter

type CloudTaskSubmitter struct {
	Client *cloudtasks.Client
}

CloudTaskSubmitter implements Submitter on top of Cloud Tasks client.

func (*CloudTaskSubmitter) CreateTask

func (s *CloudTaskSubmitter) CreateTask(ctx context.Context, req *taskspb.CreateTaskRequest, _ proto.Message) error

CreateTask creates a task, returning a gRPC status.

type CustomPayload

type CustomPayload struct {
	Method      string            // e.g. "GET" or "POST"
	RelativeURI string            // an URI relative to the task's target host
	Headers     map[string]string // HTTP headers to attach
	Body        []byte            // serialized body of the request
}

CustomPayload is returned by TaskClass's Custom, see its doc.

type Dispatcher

type Dispatcher struct {
	// Sweeper knows how to sweep transactional tasks reminders.
	//
	// If not set, Sweep calls will fail.
	Sweeper Sweeper

	// Namespace is a namespace for tasks that use DeduplicationKey.
	//
	// This is needed if two otherwise independent deployments share a single
	// Cloud Tasks instance.
	//
	// Must be valid per ValidateNamespace. Default is "".
	Namespace string

	// GAE is true when running on Appengine.
	//
	// It alters how tasks are submitted and how incoming HTTP requests are
	// authenticated.
	GAE bool

	// NoAuth can be used to disable authentication on HTTP endpoints.
	//
	// This is useful when running in development mode on localhost or in tests.
	NoAuth bool

	// CloudProject is ID of a project to use to construct full queue names.
	//
	// If not set, "default" will be used, which is pretty useless outside of
	// tests.
	CloudProject string

	// CloudRegion is a ID of a region to use to construct full queue names.
	//
	// If not set, "default" will be used, which is pretty useless outside of
	// tests.
	CloudRegion string

	// DefaultRoutingPrefix is a URL prefix for produced Cloud Tasks.
	//
	// Used only for tasks whose TaskClass doesn't provide some custom
	// RoutingPrefix.
	//
	// Default is "/internal/tasks/t/". It means generated Cloud Tasks by will
	// have target URL "/internal/tasks/t/<generated-per-task-suffix>".
	//
	// A non-default value may be valuable if you host multiple dispatchers in
	// a single process. This is a niche use case.
	DefaultRoutingPrefix string

	// DefaultTargetHost is a hostname to dispatch Cloud Tasks to by default.
	//
	// Individual task classes may override it with their own specific host.
	//
	// On GAE defaults to the GAE application itself. Elsewhere defaults to
	// "127.0.0.1", which is pretty useless outside of tests.
	DefaultTargetHost string

	// PushAs is a service account email to be used for generating OIDC tokens.
	//
	// The service account must be within the same project. The server account
	// must have "iam.serviceAccounts.actAs" permission for PushAs account.
	//
	// Optional on GAE when submitting tasks targeting GAE. Elsewhere defaults to
	// "default@example.com", which is pretty useless outside of tests.
	PushAs string

	// AuthorizedPushers is a list of service account emails to accept pushes from
	// in addition to PushAs.
	//
	// This is handy when migrating from one PushAs account to another, or when
	// submitting tasks from one service, but handing them in another.
	//
	// Optional.
	AuthorizedPushers []string

	// SweepInitiationLaunchers is a list of service account emails authorized to
	// launch sweeps via the exposed HTTP endpoint.
	SweepInitiationLaunchers []string
	// contains filtered or unexported fields
}

Dispatcher is a registry of task classes that knows how serialize and route them.

There's rarely a need to manually create instances of Dispatcher outside of Dispatcher's own tests. You should generally use the global Default dispatcher which is configured by the tq server module. Methods of the default dispatcher (such as RegisterTaskClass and AddTask) are also available as lop-level functions, prefer to use them.

The dispatcher needs a way to submit tasks to Cloud Tasks. This is the job of Submitter. It lives in the context, so that it can be mocked in tests. In production contexts (setup when using the tq server module), the submitter is initialized to be CloudTaskSubmitter. Tests will need to provide their own submitter (usually via TestingContext).

var Default Dispatcher

Default is a dispatcher installed into the server when using NewModule or NewModuleFromFlags.

The module takes care of configuring this dispatcher based on the server environment and module's options.

You still need to register your tasks in it using RegisterTaskClass.

func (*Dispatcher) AddTask

func (d *Dispatcher) AddTask(ctx context.Context, task *Task) (err error)

AddTask submits a task for later execution.

The task payload type should match some registered TaskClass. Its ID will be used to identify the task class in the serialized Cloud Tasks task body.

At some later time, in some other process, the dispatcher will invoke a handler attached to the corresponding TaskClass, based on its ID extracted from the task body.

If the given context is transactional, inherits the transaction if allowed according to the TaskClass's Kind. A transactional task will eventually be submitted to Cloud Tasks if and only if the transaction successfully commits. This requires a sweeper instance to be running somewhere, see ModuleOptions. Note that a failure to submit the task to Cloud Tasks will not abort the transaction.

If the task has a DeduplicationKey and there already was a recent task with the same TaskClass ID and DeduplicationKey, silently ignores the added task. This works only outside of transactions. Using DeduplicationKey with transactional tasks results in an error.

Annotates retriable errors with transient.Tag.

func (*Dispatcher) InstallSweepRoute

func (d *Dispatcher) InstallSweepRoute(r *router.Router, path string)

InstallSweepRoute installs a route that initiates a sweep.

It may be called periodically (e.g. by Cloud Scheduler) to launch sweeps.

func (*Dispatcher) InstallTasksRoutes

func (d *Dispatcher) InstallTasksRoutes(r *router.Router, prefix string)

InstallTasksRoutes installs tasks HTTP routes under the given prefix.

The exposed HTTP endpoints are called by Cloud Tasks service when it is time to execute a task.

func (*Dispatcher) RegisterTaskClass

func (d *Dispatcher) RegisterTaskClass(cls TaskClass) TaskClassRef

RegisterTaskClass tells the dispatcher how to route and handle tasks of some particular type.

Intended to be called during process startup. Panics if there's already a registered task class with the same ID or Prototype.

func (*Dispatcher) Sweep

func (d *Dispatcher) Sweep(ctx context.Context) error

Sweep initiates a sweep of transactional tasks reminders.

It must be called periodically (e.g. once per minute) somewhere in the fleet.

type DistributedSweeperOptions

type DistributedSweeperOptions struct {
	// SweepShards defines how many jobs to produce in each Sweep.
	//
	// Default is 16.
	SweepShards int

	// TasksPerScan caps maximum number of tasks that a sweep job will process.
	//
	// Defaults to 2048.
	TasksPerScan int

	// SecondaryScanShards caps the sharding of additional sweep scans to be
	// performed if the initial scan didn't cover the whole assigned partition.
	// In practice, this matters only when database is slow or there is a huge
	// backlog.
	//
	// Defaults to 16.
	SecondaryScanShards int

	// LessorID identifies an implementation of a system that manages leases on
	// subranges of the database.
	//
	// Default is the same ID as the database implementation ID.
	LessorID string

	// SweepTaskQueue is a Cloud Tasks queue name to use for sweep jobs.
	//
	// Can be in short or full form. See Queue in TaskClass for details. The queue
	// should be configured to allow at least 10 QPS.
	//
	// Default is "tq-sweep".
	TaskQueue string

	// SweepTaskPrefix is a URL prefix to use for sweep jobs.
	//
	// There should be a Dispatcher instance somewhere that is configured to
	// receive such tasks (via non-default ServingPrefix). This is useful if
	// you want to limit what processes process the sweeps.
	//
	// Default is "/internal/tasks".
	TaskPrefix string

	// TaskHost is a hostname to dispatch sweep jobs to.
	//
	// Default is "", meaning to use whatever is configured as default in
	// the Dispatcher.
	TaskHost string
}

DistributedSweeperOptions is configuration for the process of "sweeping" of transactional tasks reminders performed in a distributed manner using Cloud Tasks service itself to distribute work.

The sweeping process ensures all transactionally committed tasks will have a corresponding Cloud Tasks task created. It periodically scans the database for "reminder" records created whenever a task is created as part of a transaction. A reminder older than a certain age likely indicates that the corresponding AddTask call crashed right after the transaction before it had a chance to create Cloud Tasks task. For each such old reminder, the sweeping will idempotently create a Cloud Task and delete the record in the database.

DistributedSweeperOptions tune some of parameters of this process. Roughly:

1. Sweep() call in Dispatcher creates SweepShards jobs that each scan
   a portion of the database for old reminders.
2. Each such job is allowed to process no more than TasksPerScan reminder
   records. This caps its runtime and memory usage. TasksPerScan should be
   small enough so that all sweeping jobs finish before the next Sweep()
   call, but large enough so that it makes meaningful progress.
3. If a sweeping job detects there's more than TasksPerScan items it needs
   to cover, it launches SecondaryScanShards follow-up jobs that cover the
   remaining items. This should be happening in rare circumstances, only if
   the database is slow or has a large backlog.

type ExecutionInfo

type ExecutionInfo struct {
	// ExecutionCount is the total number of times that the task has received
	// a response from the handler.
	//
	// Since Cloud Tasks deletes the task once a successful response has been
	// received, all previous handler responses were failures. This number does
	// not include failures due to a lack of available instances.
	ExecutionCount int
	// contains filtered or unexported fields
}

ExecutionInfo is parsed from Cloud Tasks headers.

It is accessible from within task handlers via TaskExecutionInfo(ctx).

func TaskExecutionInfo

func TaskExecutionInfo(ctx context.Context) *ExecutionInfo

TaskExecutionInfo returns information about the currently executing task.

Returns nil if called not from a task handler.

type Handler

type Handler func(ctx context.Context, payload proto.Message) error

Handler is called to handle one enqueued task.

If the returned error is tagged with Retry tag, the request finishes with HTTP status 429, indicating to the Cloud Tasks that it should attempt to execute the task later (which it may or may not do, depending on queue's retry config). Same happens if the error is transient (i.e. tagged with the transient.Tag), except the request finishes with HTTP status 500. This difference allows to distinguish "expected" retry requests (errors tagged with Retry) from "unexpected" ones (errors tagged with transient.Tag). Retry tag should be used **only** if the handler is fully aware of Cloud Tasks retry semantics and it **explicitly** wants the task to be retried because it can't be processed right now and the handler expects that the retry may help.

For a contrived example, if the handler can process the task only after 2 PM, but it is 01:55 PM now, the handler should return an error tagged with Retry to indicate this. On the other hand, if the handler failed to process the task due to an RPC timeout or some other exceptional transient situation, it should return an error tagged with transient.Tag.

Note that it is OK (and often desirable) to tag an error with both Retry and transient.Tag. Such errors propagate through the call stack as transient, until they reach Dispatcher, which treats them as retriable.

An untagged error (or success) marks the task as "done", it won't be retried.

type InProcSweeperOptions

type InProcSweeperOptions struct {
	// SweepShards defines how many concurrent sweeping jobs to run.
	//
	// Default is 16.
	SweepShards int

	// TasksPerScan caps maximum number of tasks that a sweep job will process.
	//
	// Defaults to 2048.
	TasksPerScan int

	// SecondaryScanShards caps the sharding of additional sweep scans to be
	// performed if the initial scan didn't cover the whole assigned partition.
	// In practice, this matters only when database is slow or there is a huge
	// backlog.
	//
	// Defaults to 16.
	SecondaryScanShards int

	// SubmitBatchSize limits a single of a single processed batch.
	//
	// When processing a batch, the sweeper loads bodies of all tasks in
	// the batch, thus this setting directly affects memory usage. There will
	// be at most SubmitBatchSize*SubmitConcurrentBatches task bodies worked-on at
	// any moment in time.
	//
	// Default is 512.
	SubmitBatchSize int

	// SubmitConcurrentBatches limits how many submit batches can be worked on
	// concurrently.
	//
	// Default is 8.
	SubmitConcurrentBatches int
}

InProcSweeperOptions is configuration for the process of "sweeping" of transactional tasks reminders performed centrally in the current process.

type ModuleOptions

type ModuleOptions struct {
	// Dispatcher is a dispatcher to use.
	//
	// Default is the global Default instance.
	Dispatcher *Dispatcher

	// CloudProject is ID of a project to use to construct full queue names.
	//
	// Default is the project the server is running in.
	CloudProject string

	// CloudRegion is a ID of a region to use to construct full queue names.
	//
	// Default is the region the server is running in.
	CloudRegion string

	// Namespace is a namespace for tasks that use DeduplicationKey.
	//
	// This is needed if two otherwise independent deployments share a single
	// Cloud Tasks instance.
	//
	// Default is "".
	Namespace string

	// DefaultTargetHost is a hostname to dispatch Cloud Tasks to by default.
	//
	// Individual task classes may override it with their own specific host.
	//
	// On GAE defaults to the GAE application itself. Elsewhere has no default:
	// if the dispatcher can't figure out where to send the task, the task
	// submission fails.
	DefaultTargetHost string

	// PushAs is a service account email to be used for generating OIDC tokens.
	//
	// The service account must be within the same project. The server account
	// must have "iam.serviceAccounts.actAs" permission for `PushAs` account.
	//
	// Default is the server's own account.
	PushAs string

	// AuthorizedPushers is a list of service account emails to accept pushes from
	// in addition to PushAs.
	//
	// This is handy when migrating from one PushAs account to another, or when
	// submitting tasks from one service, but handing them in another.
	//
	// Optional.
	AuthorizedPushers []string

	// ServingPrefix is a URL path prefix to serve registered task handlers from.
	//
	// POSTs to a URL under this prefix (regardless which one) will be treated
	// as Cloud Tasks pushes.
	//
	// Default is "/internal/tasks". If set to literal "-", no routes will be
	// registered at all.
	ServingPrefix string

	// SweepMode defines how to perform sweeps of the transaction tasks reminders.
	//
	// This process is necessary to make sure all transactionally submitted tasks
	// eventually execute, even if Cloud Tasks RPCs fail. When enqueueing a task
	// the client transactionally commits a special "reminder" record, which
	// indicates an intent to submit a Cloud Task. If the subsequent Cloud Tasks
	// RPC fails (or the process crashes before attempting it), the reminder
	// record is discovered by the sweep process and used to ensure the task is
	// eventually submitted.
	//
	// There are two stages: the sweep initiation and the actual processing.
	//
	// The initiation should happen periodically and centrally: no mater how many
	// replicas of the process are running, there needs to be only one sweep
	// initiator. But it doesn't have to be the same process each time. Also
	// multiple concurrent initiations are not catastrophic, though they impose
	// huge overhead and should be avoided.
	//
	// Two ways to do sweep initiations are:
	//   * Based on a periodic external signal such as a Cloud Scheduler job or
	//     GAE cron handler. See SweepInitiationEndpoint and
	//     SweepInitiationLaunchers.
	//   * Based on a timer inside some *single* primary process. For example
	//     on Kubernetes this may be a single pod Deployment, or a zero-indexed
	//     replica in a StatefulSet. See Sweep().
	//
	// Once the initiation happens, there are two ways to process the sweep (and
	// this is what SweepMode defines):
	//   * "inproc" - do all the processing right inside the replica that
	//     performed the initiation. This has scalability and reliability limits,
	//     but it doesn't require any additional infrastructure setup and has
	//     somewhat better observability.
	//   * "distributed" - use Cloud Tasks itself to distribute the work across
	//     many replicas. This requires some configuration. See SweepTaskQueue,
	//     SweepTaskPrefix and SweepTargetHost.
	//
	// Default is "distributed" mode.
	SweepMode string

	// SweepInitiationEndpoint is a URL path that can be hit to initiate a sweep.
	//
	// GET requests to this endpoint (if they have proper authentication headers)
	// will initiate sweeps. If SweepMode is "inproc" the sweep will happen in
	// the same process that handled the request.
	//
	// On GAE default is "/internal/tasks/c/sweep". On non-GAE it is "-", meaning
	// the endpoint is not exposed. When not using the endpoint there should be
	// some single process somewhere that calls Sweep() to periodically initiate
	// sweeps.
	SweepInitiationEndpoint string

	// SweepInitiationLaunchers is a list of service account emails authorized to
	// launch sweeps via SweepInitiationEndpoint.
	//
	// Additionally on GAE the Appengine service itself is always authorized to
	// launch sweeps via cron or task queues.
	//
	// Default is the server's own account.
	SweepInitiationLaunchers []string

	// SweepTaskQueue is a Cloud Tasks queue name to use to distribute sweep
	// subtasks when running in "distributed" SweepMode.
	//
	// Can be in short or full form. See Queue in TaskClass for details. The queue
	// should be configured to allow at least 10 QPS.
	//
	// Default is "tq-sweep".
	SweepTaskQueue string

	// SweepTaskPrefix is a URL prefix to use for sweep subtasks when running
	// in "distributed" SweepMode.
	//
	// There should be a Dispatcher instance somewhere that is configured to
	// receive such tasks (via non-default ServingPrefix). This is useful if
	// you want to limit what processes process the sweeps.
	//
	// If unset defaults to the value of ServingPrefix.
	SweepTaskPrefix string

	// SweepTargetHost is a hostname to dispatch sweep subtasks to when running
	// in "distributed" SweepMode.
	//
	// This usually should be DefaultTargetHost, but it may be different if you
	// want to route sweep subtasks somewhere else.
	//
	// If unset defaults to the value of DefaultTargetHost.
	SweepTargetHost string

	// SweepShards defines how many subtasks are submitted when initiating
	// a sweep.
	//
	// It is safe to change it any time. Default is 16.
	SweepShards int
}

ModuleOptions contain configuration of the TQ server module.

It will be used to initialize Default dispatcher.

func (*ModuleOptions) Register

func (o *ModuleOptions) Register(f *flag.FlagSet)

Register registers the command line flags.

Mutates `o` by populating defaults.

type Submitter

type Submitter interface {
	// CreateTask creates a task, returning a gRPC status.
	//
	// AlreadyExists status indicates the task with request name already exists.
	// Other statuses are handled using their usual semantics.
	//
	// `msg` is the original task payload. It is non-nil only when CreateTask
	// is called on a "happy path" (i.e. not from a sweeper). This is primarily
	// useful to capture task payloads in tests. Production implementations of
	// Submitter should generally ignore it.
	//
	// Will be called from multiple goroutines at once.
	CreateTask(ctx context.Context, req *taskspb.CreateTaskRequest, msg proto.Message) error
}

Submitter is used by Dispatcher to submit Cloud Tasks.

It lives in the context, so that it can be mocked in tests. In production contexts (setup when using the tq server module), the submitter is initialized to be CloudTaskSubmitter. Tests will need to provide their own submitter (usually via TestingContext).

type Sweeper

type Sweeper interface {
	// contains filtered or unexported methods
}

Sweeper knows how sweep transaction tasks reminders.

func NewDistributedSweeper

func NewDistributedSweeper(disp *Dispatcher, opts DistributedSweeperOptions) Sweeper

NewDistributedSweeper creates a sweeper that distributes and executes sweeping tasks through the given dispatcher.

func NewInProcSweeper

func NewInProcSweeper(opts InProcSweeperOptions) Sweeper

NewInProcSweeper creates a sweeper that performs sweeping in the current process whenever Sweep is called.

type Task

type Task struct {
	// Payload is task's payload as well as indicator of its class.
	//
	// Its type will be used to find a matching registered TaskClass which defines
	// how to route and handle the task.
	Payload proto.Message

	// DeduplicationKey is optional unique key used to derive name of the task.
	//
	// If a task of a given class with a given key has already been enqueued
	// recently (within ~1h), this task will be silently ignored.
	//
	// Because there is an extra lookup cost to identify duplicate task names,
	// enqueues of named tasks have significantly increased latency.
	//
	// Named tasks can only be used outside of transactions.
	DeduplicationKey string

	// Title is optional string that identifies the task in server logs.
	//
	// It will show up as a suffix in task handler URL. It exists exclusively to
	// simplify reading server logs. It serves no other purpose! In particular,
	// it is *not* a task name.
	//
	// Handlers won't ever see it. Pass all information through the payload.
	Title string

	// Delay specifies the duration the Cloud Tasks service must wait before
	// attempting to execute the task.
	//
	// Either Delay or ETA may be set, but not both.
	Delay time.Duration

	// ETA specifies the earliest time a task may be executed.
	//
	// Either Delay or ETA may be set, but not both.
	ETA time.Time
}

Task contains task body and metadata.

type TaskClass

type TaskClass struct {
	// ID is unique identifier of this class of tasks.
	//
	// Must match `[a-zA-Z0-9_\-.]{1,100}`.
	//
	// It is used to decide how to deserialize and route the task. Changing IDs of
	// existing task classes is a disruptive operation, make sure the queue is
	// drained first. The dispatcher will permanently fail all Cloud Tasks with
	// unrecognized class IDs.
	//
	// Required.
	ID string

	// Prototype identifies a proto message type of a task payload.
	//
	// Used for its type information only. In particular it is used by AddTask
	// to discover what TaskClass matches the added task. There should be
	// one-to-one correspondence between proto message types and task classes.
	//
	// It is safe to arbitrarily change this type as long as JSONPB encoding of
	// the previous type can be decoded using the new type. The dispatcher will
	// permanently fail Cloud Tasks with bodies it can't deserialize.
	//
	// Required.
	Prototype proto.Message

	// Kind indicates whether the task requires a transaction to be enqueued.
	//
	// Note that using transactional tasks requires setting up a sweeper first,
	// see ModuleOptions.
	//
	// Default is NonTransactional which means that tasks can be enqueued only
	// outside of transactions.
	Kind TaskKind

	// Queue is a name of Cloud Tasks queue to use for the tasks.
	//
	// It can either be a short name like "default" or a full name like
	// "projects/<project>/locations/<region>/queues/<name>". If it is a full name
	// it must have the above format or RegisterTaskClass would panic.
	//
	// If it is a short queue name, the full queue name will be constructed using
	// dispatcher's CloudProject and CloudRegion if they are set.
	//
	// Required. The queue must exist already.
	Queue string

	// RoutingPrefix is a URL prefix for produced Cloud Tasks.
	//
	// Default is dispatcher's DefaultRoutingPrefix which itself defaults to
	// "/internal/tasks/t/". It means generated Cloud Tasks by default will have
	// target URL "/internal/tasks/t/<generated-per-task-suffix>".
	//
	// A non-default value can be used to route tasks of a particular class to
	// particular processes, assuming the load balancer is configured accordingly.
	RoutingPrefix string

	// TargetHost is a hostname to dispatch Cloud Tasks to.
	//
	// If unset, will use dispatcher's DefaultTargetHost.
	TargetHost string

	// Quiet, if set, instructs the dispatcher not to log bodies of tasks.
	Quiet bool

	// InheritTraceContext, if set, makes the task handler trace span be a child
	// of the span that called AddTask.
	//
	// Use it only for "one-off" tasks. Using it for deep chains of tasks usually
	// leads to messy complicated traces.
	InheritTraceContext bool

	// Custom, if given, will be called to generate a custom Cloud Tasks payload
	// (i.e. a to-be-sent HTTP request) from the task's proto payload.
	//
	// Useful for interoperability with existing code that doesn't use dispatcher.
	//
	// It is possible to customize HTTP method, relative URI, headers and the
	// request body this way. Other properties of the task (such as the target
	// host, the queue, the task name, authentication headers) are not
	// customizable.
	//
	// Such produced tasks are generally not routable by the Dispatcher. You'll
	// need to make sure there's an HTTP handler that accepts them.
	//
	// Receives the exact same context as passed to AddTask. If returns nil
	// result, the task will be dispatched as usual.
	Custom func(ctx context.Context, m proto.Message) (*CustomPayload, error)

	// Handler will be called by the dispatcher to execute the tasks.
	//
	// The handler will receive the task's payload as a proto message of the exact
	// same type as the type of Prototype. See Handler doc for more info.
	//
	// Populating this field is equivalent to calling AttachHandler after
	// registering the class. It may be left nil if the current process just wants
	// to submit tasks, but not handle them. Some other process would need to
	// attach the handler then to be able to process tasks.
	//
	// The dispatcher will permanently fail Cloud Tasks if it can't find a handler
	// for them.
	Handler Handler
}

TaskClass defines how to handles tasks of a specific proto message type.

type TaskClassRef

type TaskClassRef interface {
	// AttachHandler sets a handler which will be called by the dispatcher to
	// execute the tasks.
	//
	// The handler will receive the task's payload as a proto message of the exact
	// same type as the type of TaskClass's Prototype. See Handler doc for more
	// info.
	//
	// Panics if the class has already a handler attached.
	AttachHandler(h Handler)
}

TaskClassRef represents a TaskClass registered in a Dispatcher.

func RegisterTaskClass

func RegisterTaskClass(t TaskClass) TaskClassRef

RegisterTaskClass is a shortcut for Default.RegisterTaskClass.

type TaskKind

type TaskKind int

TaskKind describes how a task class interoperates with transactions.

const (
	// NonTransactional is a task kind for tasks that must be enqueued outside
	// of a transaction.
	NonTransactional TaskKind = 0

	// Transactional is a task kind for tasks that must be enqueued only from
	// a transaction.
	//
	// Using transactional tasks requires setting up a sweeper first, see
	// ModuleOptions.
	Transactional TaskKind = 1

	// FollowsContext is a task kind for tasks that are enqueue transactionally
	// if the context is transactional or non-transactionally otherwise.
	//
	// Using transactional tasks requires setting up a sweeper first, see
	// ModuleOptions.
	FollowsContext TaskKind = 2
)

Package Files

Documentation was rendered with GOOS=linux and GOARCH=amd64.

Jump to identifier

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to identifier