tq

Documentation

Overview

Package tq implements simple routing layer for task queue tasks.

Index

• type Dispatcher
  • func (d *Dispatcher) AddTask(c context.Context, tasks ...*Task) error
  • func (d *Dispatcher) DeleteTask(c context.Context, tasks ...*Task) error
  • func (d *Dispatcher) InstallRoutes(r *router.Router, mw router.MiddlewareChain)
  • func (d *Dispatcher) RegisterTask(prototype proto.Message, cb Handler, queue string, ...)
• type Handler
• type Task
  • func (task *Task) Name() string

Types

type Dispatcher

type Dispatcher struct {
	BaseURL string // URL prefix for all URLs, "/internal/tasks/" by default
	// contains filtered or unexported fields
}

Dispatcher submits and handles task queue tasks.

func (*Dispatcher) AddTask

func (d *Dispatcher) AddTask(c context.Context, tasks ...*Task) error

AddTask submits given tasks to an appropriate task queue.

It means, at some later time in some other GAE process, callbacks registered as handlers for corresponding proto types will be called.

If the given context is transactional or namespaced, inherits the transaction/namespace. Note if running outside of a transaction and multiple tasks are passed, the operation is not atomic: it returns an error if at least one enqueue operation failed (there's no way to figure out which one exactly).

Returns only transient errors. Unlike regular Task Queue's Add, ErrTaskAlreadyAdded is not considered an error.

func (*Dispatcher) DeleteTask

func (d *Dispatcher) DeleteTask(c context.Context, tasks ...*Task) error

DeleteTask deletes the specified tasks from their queues.

If the given context is transactional or namespaced, inherits the transaction/namespace. Note if running outside of a transaction and multiple tasks are passed, the operation is not atomic: it returns an error if at least one enqueue operation failed (there's no way to figure out which one exactly).

Returns only transient errors. Unlike regular Task Queue's Delete, attempts to delete an unknown or tombstoned task are not considered errors.

func (*Dispatcher) InstallRoutes

func (d *Dispatcher) InstallRoutes(r *router.Router, mw router.MiddlewareChain)

InstallRoutes installs appropriate HTTP routes in the router.

Must be called only after all task handlers are registered!

func (*Dispatcher) RegisterTask

func (d *Dispatcher) RegisterTask(prototype proto.Message, cb Handler, queue string, opts *taskqueue.RetryOptions)

RegisterTask tells the dispatcher that tasks of given proto type should be handled by the given handler and routed through the given task queue.

'prototype' should be a pointer to some concrete proto message. It will be used only for its type signature.

Intended to be called during process startup. Panics if such message has already been registered.

type Handler

type Handler func(c context.Context, payload proto.Message, execCount int) error

Handler is called to handle one enqueued task.

The passed context is produced by a middleware chain installed with InstallHandlers.

execCount corresponds to X-AppEngine-TaskExecutionCount header value: it is 1 on first execution attempt, 2 on a retry, and so on.

May return transient errors. In this case, task queue may attempt to redeliver the task (depending on RetryOptions).

A fatal error (or success) mark the task as "done", it won't be retried.

type Task

type Task struct {
	// Payload is task's payload as well as indicator of its type.
	//
	// Tasks are routed based on type of the payload message, see RegisterTask.
	Payload proto.Message

	// NamePrefix, if not empty, is a string that will be prefixed to the task's
	// name. Characters in NamePrefix must be appropriate task queue name
	// characters. NamePrefix can be useful because the Task Queue system allows
	// users to search for tasks by prefix.
	//
	// Lexicographically close names can cause hot spots in the Task Queues
	// backend. If NamePrefix is specified, users should try and ensure that
	// it is friendly to sharding (e.g., begins with a hash string).
	//
	// Setting NamePrefix and/or DeduplicationKey will result in a named task
	// being generated. This task can be cancelled using DeleteTask.
	NamePrefix string

	// DeduplicationKey is optional unique key of the task.
	//
	// If a task of a given proto type with a given key has already been enqueued
	// recently, this task will be silently ignored.
	//
	// Such tasks can only be used outside of transactions.
	//
	// Setting NamePrefix and/or DeduplicationKey will result in a named task
	// being generated. This task can be cancelled using DeleteTask.
	DeduplicationKey string

	// Title is optional string that identifies the task in HTTP logs.
	//
	// It will show up as a suffix in task handler URL. It exists exclusively to
	// simplify reading HTTP 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 task body.
	Title string

	// Delay specifies the duration the task queue service must wait before
	// executing 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

	// Retry options for this task.
	//
	// If given, overrides default options set when this task was registered.
	RetryOptions *taskqueue.RetryOptions
}

Task contains task body and additional parameters that influence how it is routed.

func (*Task) Name

func (task *Task) Name() string

Name generates and returns the task's name.

If the task is not a named task (doesn't have NamePrefix or DeduplicationKey set), this will return an empty string.