protocol

Documentation

Overview

Package protocol provides means to encode/decode task messages as described in https://github.com/celery/celery/blob/master/docs/internals/protocol.rst.

Index

• Constants
• type JSONSerializer
  • func NewJSONSerializer() *JSONSerializer
  • func (ser *JSONSerializer) Decode(p int, s string, t *Task) error
  • func (ser *JSONSerializer) Encode(p int, t *Task) (s string, err error)
• type Serializer
• type SerializerRegistry
  • func NewSerializerRegistry() *SerializerRegistry
  • func (r *SerializerRegistry) Decode(raw []byte) (*Task, error)
  • func (r *SerializerRegistry) Encode(queue, mime string, prot int, t *Task) ([]byte, error)
  • func (r *SerializerRegistry) Register(serializer Serializer, mime, encoding string)
• type Task
  • func (t *Task) IsExpired() bool

Constants

const (
	V1 = 1
	V2 = 2
)

Supported protocol versions.

const (
	MimeJSON = "application/json"
)

The mime-type describing the serializers.

Types

type JSONSerializer

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

JSONSerializer encodes/decodes a task messages in JSON format. The zero value is not usable.

func NewJSONSerializer

func NewJSONSerializer() *JSONSerializer

NewJSONSerializer returns JSONSerializer.

func (*JSONSerializer) Decode

func (ser *JSONSerializer) Decode(p int, s string, t *Task) error

Decode parses the JSON-encoded message body s depending on Celery protocol version (v1 or v2). The task t is updated with the decoded params.

func (*JSONSerializer) Encode

func (ser *JSONSerializer) Encode(p int, t *Task) (s string, err error)

Encode encodes task t using protocol version p and returns the message body s.

type Serializer

type Serializer interface {
	// Decode decodes the message body s into task t
	// using protocol p which could be version 1 or 2.
	Decode(p int, s string, t *Task) error
	// Encode encodes task t using protocol p and returns the message body s.
	Encode(p int, t *Task) (s string, err error)
}

Serializer encodes/decodes Celery tasks (message's body param to be precise). See https://docs.celeryq.dev/projects/kombu/en/latest/userguide/serialization.html.

type SerializerRegistry

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

SerializerRegistry encodes/decodes task messages using registered serializers. Celery relies on JSON format to store message metadata such as content type and headers. Task details (args, kwargs) are encoded in message body in base64 and JSON by default. The encoding is indicated by body_encoding and content-type message params. Therefore a client doesn't have to specify the formats since the registry can pick an appropriate decoder based on the aforementioned params.

func NewSerializerRegistry

func NewSerializerRegistry() *SerializerRegistry

NewSerializerRegistry creates a registry of serializers.

func (*SerializerRegistry) Decode

func (r *SerializerRegistry) Decode(raw []byte) (*Task, error)

Decode decodes the raw message and returns a task info. If the header doesn't contain a task name, then protocol v1 is assumed. Otherwise the protocol v2 is used.

func (*SerializerRegistry) Encode

func (r *SerializerRegistry) Encode(queue, mime string, prot int, t *Task) ([]byte, error)

Encode encodes the task message.

func (*SerializerRegistry) Register

func (r *SerializerRegistry) Register(serializer Serializer, mime, encoding string)

Register registers a custom serializer where mime is the mime-type describing the serialized structure, e.g., application/json, and encoding is the content encoding which is usually utf-8 or binary.

type Task

type Task struct {
	// ID id a unique id of the task in UUID v4 format (required).
	ID string
	// Name is a name of the task (required).
	Name string
	// Args is a list of arguments.
	// It will be an empty list if not provided.
	Args []interface{}
	// Kwargs is a dictionary of keyword arguments.
	// It will be an empty dictionary if not provided.
	Kwargs map[string]interface{}
	// Expires is an expiration date in ISO 8601 format.
	// If not provided the message will never expire.
	// The message will be expired when the message is received and the expiration date has been exceeded.
	Expires time.Time
}

Task represents a task message that provides essential params to run a task.

func (*Task) IsExpired

func (t *Task) IsExpired() bool

IsExpired returns true if the message is expired and shouldn't be processed.