celery

README

celery

celery

Documentation

Index

• Constants
• Variables
• type Celery
  • func (c Celery) Delay(task string, args ...any) (*Result, error)
  • func (c Celery) DelayWithQueue(queue string, task string, args ...any) (*Result, error)
• type CeleryBackend
• type CeleryBody
  • func (cb CeleryBody) ArgsRepr() string
  • func (cb CeleryBody) Encode() string
• type CeleryBodyData
• type CeleryBroker
• type CeleryDeliveryInfo
• type CeleryHeaders
• type CeleryMessage
  • func NewCeleryMessage(queue string, task string, args ...any) *CeleryMessage
• type CeleryProperties
• type RedisBackend
  • func NewRedisBackend(url string) *RedisBackend
  • func (r RedisBackend) GetCeleryResult(ID string) (*ResultMessage, error)
• type RedisBroker
  • func NewRedisBroker(url string) *RedisBroker
  • func (r RedisBroker) SendCeleryMessage(queue string, cm *CeleryMessage) error
• type Result
  • func (r *Result) Get() error
  • func (r *Result) Ready() bool
  • func (r *Result) Success() bool
  • func (r *Result) Wait(timeout time.Duration) bool
• type ResultMessage

Constants

const (
	REDIS_ID_PREFIX   = "celery-task-meta-"
	REDIS_TIME_LAYOUT = "2006-01-02T15:04:05.999999"
)

Variables

var (
	READY_STATES = mapset.NewSet("SUCCESS", "FAILURE", "REVOKED")
)

Types

type Celery

type Celery struct {
	Broker  CeleryBroker
	Backend CeleryBackend
}

func (Celery) Delay

func (c Celery) Delay(task string, args ...any) (*Result, error)

func (Celery) DelayWithQueue

func (c Celery) DelayWithQueue(queue string, task string, args ...any) (*Result, error)

type CeleryBackend

type CeleryBackend interface {
	GetCeleryResult(string) (*ResultMessage, error)
}

type CeleryBody

type CeleryBody struct {
	Args   []any          `json:"args"`
	Kwargs map[string]any `json:"kwargs"`
	Data   CeleryBodyData `json:"data"`
}

func (CeleryBody) ArgsRepr

func (cb CeleryBody) ArgsRepr() string

func (CeleryBody) Encode

func (cb CeleryBody) Encode() string

type CeleryBodyData

type CeleryBodyData struct {
	Callbacks []string `json:"callbacks"`
	Errbacks  []string `json:"errbacks"`
	Chain     []string `json:"chain"`
	Chord     *string  `json:"chord"`
}

type CeleryBroker

type CeleryBroker interface {
	SendCeleryMessage(string, *CeleryMessage) error
}

type CeleryDeliveryInfo

type CeleryDeliveryInfo struct {
	Exchange   string `json:"exchange"`
	RoutingKey string `json:"routing_key"`
}

type CeleryHeaders

type CeleryHeaders struct {
	Lang         string    `json:"lang"`
	Task         string    `json:"task"`
	ID           string    `json:"id"`
	Shadow       *string   `json:"shadow"`
	Eta          *string   `json:"eta"`
	Expires      *string   `json:"expires"`
	Group        *int      `json:"group"`
	GroupIndex   *int      `json:"group_index"`
	Retries      int       `json:"retries"`
	TimeLimit    []*string `json:"timelimit"`
	RootID       string    `json:"root_id"`
	ParentID     *int      `json:"parent_id"`
	ArgsRepr     string    `json:"argsrepr"`
	KwargsRepr   string    `json:"kwargsrepr"`
	Origin       string    `json:"origin"`
	IgnoreResult bool      `json:"ignore_result"`
}

type CeleryMessage

type CeleryMessage struct {
	Body            string           `json:"body"`
	ContentEncoding string           `json:"content-encoding"`
	ContentType     string           `json:"content-type"`
	Headers         CeleryHeaders    `json:"headers"`
	Properties      CeleryProperties `json:"properties"`
}

func NewCeleryMessage

func NewCeleryMessage(queue string, task string, args ...any) *CeleryMessage

type CeleryProperties

type CeleryProperties struct {
	CorrelationID string             `json:"correlation_id"`
	ReplyTo       string             `json:"reply_to"`
	DeliveryMode  int                `json:"delivery_mode"`
	DeliveryInfo  CeleryDeliveryInfo `json:"delivery_info"`
	Priority      int                `json:"priority"`
	BodyEncoding  string             `json:"body_encoding"`
	DeliveryTag   string             `json:"delivery_tag"`
}

type RedisBackend

type RedisBackend struct {
	Conn *redis.Client
	Ctx  context.Context
}

func NewRedisBackend

func NewRedisBackend(url string) *RedisBackend

func (RedisBackend) GetCeleryResult

func (r RedisBackend) GetCeleryResult(ID string) (*ResultMessage, error)

type RedisBroker

type RedisBroker struct {
	Conn *redis.Client
	Ctx  context.Context
}

func NewRedisBroker

func NewRedisBroker(url string) *RedisBroker

func (RedisBroker) SendCeleryMessage

func (r RedisBroker) SendCeleryMessage(queue string, cm *CeleryMessage) error

type Result

type Result struct {
	ID      string
	Backend CeleryBackend
	Result  *ResultMessage
}

func (*Result) Get

func (r *Result) Get() error

func (*Result) Ready

func (r *Result) Ready() bool

func (*Result) Success

func (r *Result) Success() bool

func (*Result) Wait

func (r *Result) Wait(timeout time.Duration) bool

type ResultMessage

type ResultMessage struct {
	ID        string     `json:"task_id"`
	Status    string     `json:"status"`
	Result    any        `json:"result"`
	Traceback any        `json:"traceback"`
	Children  []any      `json:"children"`
	DateDone  *time.Time `json:"date_done"`
}