queue

package
v0.11.0 Latest Latest
Warning

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

 Go to latest Published: Jul 6, 2026 License: MIT Imports: 23 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	SignatureHeader = "x-gogo-signature"
	TimestampHeader = "x-gogo-timestamp"
	KeyIDHeader     = "x-gogo-key-id"
	RedactedValue   = "[REDACTED]"
)

Variables

View Source 
var (
	ErrQueueEmpty   = errors.New("queue empty")
	ErrBrokerClosed = errors.New("broker closed")
)
View Source 
var (
	ErrRetryRequested = errors.New("retry requested")
	ErrSoftTimeout    = errors.New("soft timeout exceeded")
	ErrHardTimeout    = errors.New("hard timeout exceeded")
)
View Source 
var (
	ErrMessageSigningKey       = errors.New("message signing key error")
	ErrInvalidMessageSignature = errors.New("invalid message signature")
	ErrMessageExpired          = errors.New("message timestamp outside replay window")
	ErrRejectedContentType     = errors.New("rejected content type")
	ErrInvalidBrokerTLS        = errors.New("invalid broker tls configuration")
)
View Source 
var (
	ErrUnknownSerializer      = errors.New("unknown serializer")
	ErrUntrustedSerializer    = errors.New("untrusted serializer")
	ErrUnsupportedCompression = errors.New("unsupported compression")
)
View Source 
var (
	ErrDuplicateTask = errors.New("duplicate task")
	ErrInvalidTask   = errors.New("invalid task")
)
View Source 
var (
	ErrWorkerNotConfigured = errors.New("worker not configured")
	ErrWorkerRunning       = errors.New("worker already running")
	ErrWorkerStopped       = errors.New("worker stopped")
	ErrWorkerMemoryLimit   = errors.New("worker memory limit exceeded")
	ErrTaskNotRegistered   = errors.New("task not registered")
)
View Source 
var ErrScheduleLocked = errors.New("schedule locked")
View Source 
var ErrUnsupportedRuntimeURL = errors.New("unsupported queue runtime URL")

Functions

func CanRetry 

func CanRetry(options TaskOptions, currentRetries int) bool

func CanTransition 

func CanTransition(from State, to State) bool

func ComputeRetryDelay 

func ComputeRetryDelay(options TaskOptions, currentRetries int, jitter func(time.Duration) time.Duration) time.Duration

func QueueAdminActions 

func QueueAdminActions(options QueueAdminOptions) []admin.Action

func RegisterAdmin 

func RegisterAdmin(registry *admin.Registry, options QueueAdminOptions) error

func RegisterBrokerFactory 

func RegisterBrokerFactory(scheme string, factory BrokerFactory)

func RegisterResultBackendFactory 

func RegisterResultBackendFactory(scheme string, factory ResultBackendFactory)

func RegisterScheduleStoreFactory 

func RegisterScheduleStoreFactory(scheme string, factory ScheduleStoreFactory)

func Retry 

func Retry(err error, options ...RetryOption) error

func ValidateBrokerTLS 

func ValidateBrokerTLS(config BrokerTLSConfig) error

Types

type AckPolicy 

type AckPolicy string

AckPolicy controls when workers acknowledge broker deliveries. 

const (
	AckEarly  AckPolicy = "early"
	AckLate   AckPolicy = "late"
	AckManual AckPolicy = "manual"
)

type ActiveTask 

type ActiveTask struct {
	ID        string
	Name      string
	Queue     string
	Hostname  string
	StartedAt time.Time
}

type App 

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

App owns task registration and lookup.

func NewApp 

func NewApp(options AppOptions) *App

func (*App) DiscoverTasks 

func (a *App) DiscoverTasks(providers ...TaskProvider) error

func (*App) RegisterTask 

func (a *App) RegisterTask(name string, fn TaskFunc, options TaskOptions) (Task, error)

func (*App) SendTask 

func (a *App) SendTask(ctx context.Context, broker Broker, signature Signature, options SendOptions) (BrokerMessage, error)

func (*App) SetTaskRateLimit 

func (a *App) SetTaskRateLimit(name string, limit RateLimit) error

func (*App) SetTaskTimeLimit 

func (a *App) SetTaskTimeLimit(name string, soft time.Duration, hard time.Duration) error

func (*App) Task 

func (a *App) Task(name string) (Task, bool)

func (*App) Tasks 

func (a *App) Tasks() []Task

type AppOptions 

type AppOptions struct {
	DefaultQueue      string
	DefaultSerializer string
	DefaultMaxRetries int
	DefaultAckPolicy  AckPolicy
}

AppOptions configures the queue app defaults.

type AutoscaleConfig 

type AutoscaleConfig struct {
	MinConcurrency    int
	MaxConcurrency    int
	ScaleUpReadyTasks int
	ScaleDownIdleFor  time.Duration
}

type AutoscaleState 

type AutoscaleState struct {
	MinConcurrency    int
	MaxConcurrency    int
	ScaleUpReadyTasks int
	ScaleDownIdleFor  time.Duration
}

func ResolveAutoscale 

func ResolveAutoscale(baseConcurrency int, config AutoscaleConfig) AutoscaleState

func (AutoscaleState) Target 

func (s AutoscaleState) Target(readyTasks int) int

type Beat 

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

func NewBeat 

func NewBeat(app *App, broker Broker, store ScheduleStore, options BeatOptions) *Beat

func (*Beat) Tick 

func (b *Beat) Tick(ctx context.Context) (int, error)

type BeatOptions 

type BeatOptions struct {
	Router  *Router
	Now     func() time.Time
	LockTTL time.Duration
}

type Broker 

type Broker interface {
	Publish(context.Context, string, Envelope, BrokerPublishOptions) (BrokerMessage, error)
	Consume(context.Context, string, BrokerConsumeOptions) (BrokerMessage, error)
	Ack(context.Context, BrokerMessage) error
	Nack(context.Context, BrokerMessage, bool) error
	Requeue(context.Context, BrokerMessage, time.Duration) error
	DeclareQueue(context.Context, string, BrokerQueueOptions) error
	PurgeQueue(context.Context, string) (int, error)
	InspectQueues(context.Context) ([]BrokerQueueInfo, error)
	Close() error
}

Broker is the stable worker-facing queue broker contract.

func NewBrokerFromURL 

func NewBrokerFromURL(config RuntimeConfig) (Broker, error)

type BrokerConsumeOptions 

type BrokerConsumeOptions struct {
	VisibilityTimeout time.Duration
}

type BrokerFactory 

type BrokerFactory func(RuntimeConfig) (Broker, error)

type BrokerMessage 

type BrokerMessage struct {
	DeliveryID string
	Queue      string
	Envelope   Envelope
	Priority   int
	Attempts   int
	VisibleAt  time.Time
	Deadline   time.Time
}

type BrokerPublishOptions 

type BrokerPublishOptions struct {
	Priority   int
	RoutingKey string
	Headers    map[string]string
}

type BrokerQueueInfo 

type BrokerQueueInfo struct {
	Name     string
	Ready    int
	InFlight int
	Durable  bool
}

type BrokerQueueOptions 

type BrokerQueueOptions struct {
	Durable           bool
	VisibilityTimeout time.Duration
}

type BrokerTLSConfig 

type BrokerTLSConfig struct {
	URL                string
	TLSEnabled         bool
	ServerName         string
	InsecureSkipVerify bool
}

type ClockedSchedule 

type ClockedSchedule struct {
	RunAt time.Time
}

func (ClockedSchedule) Next 

func (s ClockedSchedule) Next(after time.Time) (time.Time, bool)

func (ClockedSchedule) NextRun 

func (s ClockedSchedule) NextRun(lastRunAt *time.Time, now time.Time) (time.Time, bool)

type Compression 

type Compression string
const (
	CompressionNone Compression = "none"
	CompressionGzip Compression = "gzip"
	CompressionZstd Compression = "zstd"
)

type ContentTypeAllowlist 

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

func NewContentTypeAllowlist 

func NewContentTypeAllowlist(contentTypes ...string) ContentTypeAllowlist

func (ContentTypeAllowlist) Validate 

func (a ContentTypeAllowlist) Validate(contentType string) error

type CrontabSchedule 

type CrontabSchedule struct {
	Minute     string
	Hour       string
	DayOfMonth string
	Month      string
	DayOfWeek  string
	Location   *time.Location
}

func (CrontabSchedule) Next 

func (s CrontabSchedule) Next(after time.Time) (time.Time, bool)

func (CrontabSchedule) NextRun 

func (s CrontabSchedule) NextRun(lastRunAt *time.Time, now time.Time) (time.Time, bool)

type Envelope 

type Envelope struct {
	ID            string            `json:"id"`
	RootID        string            `json:"root_id,omitempty"`
	ParentID      string            `json:"parent_id,omitempty"`
	GroupID       string            `json:"group_id,omitempty"`
	ChordID       string            `json:"chord_id,omitempty"`
	Name          string            `json:"name"`
	Args          []any             `json:"args,omitempty"`
	Kwargs        map[string]any    `json:"kwargs,omitempty"`
	Headers       map[string]string `json:"headers,omitempty"`
	Retries       int               `json:"retries"`
	ETA           *time.Time        `json:"eta,omitempty"`
	Expires       *time.Time        `json:"expires,omitempty"`
	Queue         string            `json:"queue,omitempty"`
	Priority      int               `json:"priority,omitempty"`
	ReplyTo       string            `json:"reply_to,omitempty"`
	CorrelationID string            `json:"correlation_id,omitempty"`
	CreatedAt     time.Time         `json:"created_at"`
}

Envelope is the durable broker message for one task.

func NewEnvelope 

func NewEnvelope(signature Signature, options EnvelopeOptions) Envelope

type EnvelopeOptions 

type EnvelopeOptions struct {
	ID            string
	RootID        string
	ParentID      string
	GroupID       string
	ChordID       string
	Retries       int
	ReplyTo       string
	CorrelationID string
	CreatedAt     time.Time
}

type Event 

type Event struct {
	Type     EventType
	Hostname string
	TaskID   string
	TaskName string
	Queue    string
	State    State
	Error    string
	At       time.Time
	Fields   map[string]any
}

type EventRecorder 

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

func NewEventRecorder 

func NewEventRecorder() *EventRecorder

func (*EventRecorder) Clear 

func (r *EventRecorder) Clear()

func (*EventRecorder) Disable 

func (r *EventRecorder) Disable()

func (*EventRecorder) EmitQueueEvent 

func (r *EventRecorder) EmitQueueEvent(_ context.Context, event Event)

func (*EventRecorder) Enable 

func (r *EventRecorder) Enable()

func (*EventRecorder) Enabled 

func (r *EventRecorder) Enabled() bool

func (*EventRecorder) Events 

func (r *EventRecorder) Events() []Event

type EventSink 

type EventSink interface {
	EmitQueueEvent(context.Context, Event)
}

type EventType 

type EventType string
const (
	EventWorkerOnline    EventType = "worker.online"
	EventWorkerHeartbeat EventType = "worker.heartbeat"
	EventWorkerOffline   EventType = "worker.offline"
	EventTaskSent        EventType = "task.sent"
	EventTaskReceived    EventType = "task.received"
	EventTaskStarted     EventType = "task.started"
	EventTaskSucceeded   EventType = "task.succeeded"
	EventTaskFailed      EventType = "task.failed"
	EventTaskRetried     EventType = "task.retried"
	EventTaskRevoked     EventType = "task.revoked"
)

type GoroutinePool 

type GoroutinePool struct{}

func NewGoroutinePool 

func NewGoroutinePool() *GoroutinePool

func (*GoroutinePool) Close 

func (p *GoroutinePool) Close(context.Context) error

func (*GoroutinePool) Run 

func (p *GoroutinePool) Run(ctx context.Context, executable PoolExecutable) (any, error)

func (*GoroutinePool) Strategy 

func (p *GoroutinePool) Strategy() PoolStrategy

type GroupResult 

type GroupResult struct {
	ID        string
	Children  []string
	CreatedAt time.Time
}

GroupResult stores a group result handle.

func (GroupResult) Clone 

func (g GroupResult) Clone() GroupResult

type InspectOptions 

type InspectOptions struct {
	App         *App
	Broker      Broker
	Store       ScheduleStore
	Workers     []*Worker
	Revocations *RevocationRegistry
	Events      *EventRecorder
}

type InspectReport 

type InspectReport struct {
	Registered []Task
	Active     []ActiveTask
	Scheduled  []ScheduleEntry
	Reserved   []BrokerMessage
	Queues     []BrokerQueueInfo
	Workers    []WorkerStats
}

type Inspector 

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

func NewInspector 

func NewInspector(options InspectOptions) *Inspector

func (*Inspector) ActiveTasks 

func (i *Inspector) ActiveTasks() []ActiveTask

func (*Inspector) DisableEvents 

func (i *Inspector) DisableEvents()

func (*Inspector) EnableEvents 

func (i *Inspector) EnableEvents()

func (*Inspector) Ping 

func (i *Inspector) Ping(context.Context) PingResponse

func (*Inspector) PoolGrow 

func (i *Inspector) PoolGrow(worker *Worker, delta int)

func (*Inspector) PoolRestart 

func (i *Inspector) PoolRestart(ctx context.Context, worker *Worker) error

func (*Inspector) PoolShrink 

func (i *Inspector) PoolShrink(worker *Worker, delta int)

func (*Inspector) QueueLengths 

func (i *Inspector) QueueLengths(ctx context.Context) ([]BrokerQueueInfo, error)

func (*Inspector) RateLimit 

func (i *Inspector) RateLimit(taskName string, limit RateLimit) error

func (*Inspector) RegisteredTasks 

func (i *Inspector) RegisteredTasks() []Task

func (*Inspector) Report 

func (i *Inspector) Report(ctx context.Context) (InspectReport, error)

func (*Inspector) ReservedTasks 

func (i *Inspector) ReservedTasks(context.Context) ([]BrokerMessage, error)

func (*Inspector) RevokeByStampedHeaders 

func (i *Inspector) RevokeByStampedHeaders(name string, value string)

func (*Inspector) RevokeTask 

func (i *Inspector) RevokeTask(taskID string)

func (*Inspector) ScheduledTasks 

func (i *Inspector) ScheduledTasks(ctx context.Context) ([]ScheduleEntry, error)

func (*Inspector) Shutdown 

func (i *Inspector) Shutdown(ctx context.Context, worker *Worker, mode ShutdownMode) error

func (*Inspector) TimeLimit 

func (i *Inspector) TimeLimit(taskName string, soft time.Duration, hard time.Duration) error

func (*Inspector) WorkerStats 

func (i *Inspector) WorkerStats() []WorkerStats

type IntervalSchedule 

type IntervalSchedule struct {
	Every   time.Duration
	StartAt time.Time
}

func (IntervalSchedule) Next 

func (s IntervalSchedule) Next(after time.Time) (time.Time, bool)

func (IntervalSchedule) NextRun 

func (s IntervalSchedule) NextRun(lastRunAt *time.Time, now time.Time) (time.Time, bool)

type MemoryScheduleStore 

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

func NewMemoryScheduleStore 

func NewMemoryScheduleStore(options MemoryScheduleStoreOptions) *MemoryScheduleStore

func (*MemoryScheduleStore) List 

func (s *MemoryScheduleStore) List(context.Context) ([]ScheduleEntry, error)

func (*MemoryScheduleStore) Lock 

func (s *MemoryScheduleStore) Lock(_ context.Context, name string, ttl time.Duration) (ScheduleLock, error)

func (*MemoryScheduleStore) Save 

func (s *MemoryScheduleStore) Save(_ context.Context, entry ScheduleEntry) error

type MemoryScheduleStoreOptions 

type MemoryScheduleStoreOptions struct {
	Now func() time.Time
}

type MessageSigner 

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

func NewMessageSigner 

func NewMessageSigner(options MessageSignerOptions) *MessageSigner

func (*MessageSigner) Sign 

func (s *MessageSigner) Sign(envelope Envelope) (map[string]string, error)

func (*MessageSigner) Verify 

func (s *MessageSigner) Verify(envelope Envelope, headers map[string]string) error

type MessageSignerOptions 

type MessageSignerOptions struct {
	PrimaryKeyID string
	Keys         map[string][]byte
	Now          func() time.Time
	ReplayWindow time.Duration
}

type NextSchedule 

type NextSchedule interface {
	Next(after time.Time) (time.Time, bool)
}

type Payload 

type Payload struct {
	Serializer  string
	ContentType string
	Compression Compression
	Body        []byte
}

Payload is a serialized queue message body.

type PingResponse 

type PingResponse struct {
	OK       bool
	Hostname string
	At       time.Time
}

type Pool 

type Pool interface {
	Strategy() PoolStrategy
	Run(context.Context, PoolExecutable) (any, error)
	Close(context.Context) error
}

Pool executes task callables behind a stable worker runtime boundary.

type PoolExecutable 

type PoolExecutable func(context.Context) (any, error)

type PoolStrategy 

type PoolStrategy string
const (
	PoolGoroutine     PoolStrategy = "goroutine"
	PoolSolo          PoolStrategy = "solo"
	PoolProcessBacked PoolStrategy = "process"
)

type ProcessPool 

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

ProcessPool is a process-backed execution boundary where supported by the host. The initial implementation preserves the boundary contract while executing through goroutines so callers can opt into the strategy without a platform fork.

func NewProcessPool 

func NewProcessPool(ProcessPoolOptions) *ProcessPool

func (*ProcessPool) Close 

func (p *ProcessPool) Close(ctx context.Context) error

func (*ProcessPool) Run 

func (p *ProcessPool) Run(ctx context.Context, executable PoolExecutable) (any, error)

func (*ProcessPool) Strategy 

func (p *ProcessPool) Strategy() PoolStrategy

type ProcessPoolOptions 

type ProcessPoolOptions struct{}

type QueueAdminModel 

type QueueAdminModel struct {
	Metadata models.Metadata
	Admin    admin.ModelAdmin
}

func QueueAdminModels 

func QueueAdminModels(options QueueAdminOptions) []QueueAdminModel

type QueueAdminOptions 

type QueueAdminOptions struct {
	Broker      Broker
	Store       ScheduleStore
	Revocations *RevocationRegistry
	Inspector   *Inspector
}

type QueueAdminView 

type QueueAdminView struct {
	Name       string
	ModelLabel string
	Columns    []string
	Actions    []string
}

func QueueAdminViews 

func QueueAdminViews(options QueueAdminOptions) []QueueAdminView

type RateLimit 

type RateLimit struct {
	Limit  int
	Period time.Duration
}

RateLimit describes a per-task execution limit.

type RateLimiter 

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

func NewRateLimiter 

func NewRateLimiter(options RateLimiterOptions) *RateLimiter

func (*RateLimiter) Allow 

func (l *RateLimiter) Allow(taskName string, limit RateLimit) (bool, time.Duration)

func (*RateLimiter) Reset 

func (l *RateLimiter) Reset(taskName string)

type RateLimiterOptions 

type RateLimiterOptions struct {
	Now func() time.Time
}

type Redactor 

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

func NewRedactor 

func NewRedactor(options RedactorOptions) Redactor

func (Redactor) RedactEnvelope 

func (r Redactor) RedactEnvelope(envelope Envelope) Envelope

func (Redactor) RedactEvent 

func (r Redactor) RedactEvent(event Event) Event

func (Redactor) RedactResult 

func (r Redactor) RedactResult(result Result) Result

type RedactorOptions 

type RedactorOptions struct {
	SensitiveKeys []string
}

type Result 

type Result struct {
	TaskID    string
	State     State
	Result    any
	Error     string
	Traceback string
	Children  []string
	ExpiresAt *time.Time
	CreatedAt time.Time
	UpdatedAt time.Time
}

Result stores task execution state and payloads.

func (Result) Clone 

func (r Result) Clone() Result

type ResultBackend 

type ResultBackend interface {
	StoreResult(context.Context, Result) error
	GetResult(context.Context, string) (Result, error)
	Forget(context.Context, string) error
	Wait(context.Context, string, time.Duration) (Result, error)
	Children(context.Context, string) ([]string, error)
	GroupResult(context.Context, string, []string) (GroupResult, error)
	ChordCounter(context.Context, string, int) (int, error)
}

ResultBackend is the worker-facing result storage contract.

func NewResultBackendFromURL 

func NewResultBackendFromURL(config RuntimeConfig) (ResultBackend, error)

type ResultBackendFactory 

type ResultBackendFactory func(RuntimeConfig) (ResultBackend, error)

type RetryError 

type RetryError struct {
	Err        error
	Countdown  time.Duration
	ETA        *time.Time
	MaxRetries *int
}

func AsRetry 

func AsRetry(err error) (*RetryError, bool)

func (*RetryError) Error 

func (e *RetryError) Error() string

func (*RetryError) Unwrap 

func (e *RetryError) Unwrap() error

type RetryOption 

type RetryOption func(*RetryError)

func RetryCountdown 

func RetryCountdown(countdown time.Duration) RetryOption

func RetryETA 

func RetryETA(eta time.Time) RetryOption

func RetryMaxRetries 

func RetryMaxRetries(maxRetries int) RetryOption

type RevocationRegistry 

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

func NewRevocationRegistry 

func NewRevocationRegistry() *RevocationRegistry

func (*RevocationRegistry) ClearStampedHeader 

func (r *RevocationRegistry) ClearStampedHeader(name string, value string)

func (*RevocationRegistry) ClearTask 

func (r *RevocationRegistry) ClearTask(taskID string)

func (*RevocationRegistry) IsRevoked 

func (r *RevocationRegistry) IsRevoked(envelope Envelope) bool

func (*RevocationRegistry) RevokeStampedHeader 

func (r *RevocationRegistry) RevokeStampedHeader(name string, value string)

func (*RevocationRegistry) RevokeTask 

func (r *RevocationRegistry) RevokeTask(taskID string)

type Route 

type Route struct {
	Queue      string
	RoutingKey string
	Priority   int
	Headers    map[string]string
}

type RouteFunc 

type RouteFunc func(Signature) (Route, bool)

type Router 

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

func NewRouter 

func NewRouter(options RouterOptions) *Router

func (*Router) Route 

func (r *Router) Route(signature Signature, task Task) Route

type RouterOptions 

type RouterOptions struct {
	DefaultQueue   string
	StaticRoutes   map[string]Route
	DynamicRoutes  []RouteFunc
	DefaultHeaders map[string]string
}

type RuntimeConfig 

type RuntimeConfig struct {
	BrokerURL     string
	ResultBackend string
	ScheduleStore string
}

type Schedule 

type Schedule interface {
	NextRun(lastRunAt *time.Time, now time.Time) (time.Time, bool)
}

type ScheduleEntry 

type ScheduleEntry struct {
	Name          string
	Signature     Signature
	Schedule      Schedule
	Enabled       bool
	OneOff        bool
	LastRunAt     *time.Time
	TotalRunCount int
	Send          SendOptions
}

func (ScheduleEntry) Clone 

func (e ScheduleEntry) Clone() ScheduleEntry

type ScheduleLock 

type ScheduleLock interface {
	Release(context.Context) error
}

type ScheduleStore 

type ScheduleStore interface {
	List(context.Context) ([]ScheduleEntry, error)
	Save(context.Context, ScheduleEntry) error
	Lock(context.Context, string, time.Duration) (ScheduleLock, error)
}

func NewScheduleStoreFromURL 

func NewScheduleStoreFromURL(config RuntimeConfig) (ScheduleStore, error)

type ScheduleStoreFactory 

type ScheduleStoreFactory func(RuntimeConfig) (ScheduleStore, error)

type SendOptions 

type SendOptions struct {
	Router        *Router
	ID            string
	RootID        string
	ParentID      string
	GroupID       string
	ChordID       string
	Retries       int
	ReplyTo       string
	CorrelationID string
	CreatedAt     time.Time
	Events        EventSink
}

type SensitiveValue 

type SensitiveValue struct {
	Value any
}

func Sensitive 

func Sensitive(value any) SensitiveValue

type SerializationOptions 

type SerializationOptions struct {
	AllowUntrustedSerializers []string
}

type SerializationRegistry 

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

SerializationRegistry stores serializers and trust settings.

func NewSerializationRegistry 

func NewSerializationRegistry(options SerializationOptions) *SerializationRegistry

func (*SerializationRegistry) Decode 

func (r *SerializationRegistry) Decode(payload Payload, value any) error

func (*SerializationRegistry) Encode 

func (r *SerializationRegistry) Encode(serializerName string, value any, compression Compression) (Payload, error)

func (*SerializationRegistry) Register 

func (r *SerializationRegistry) Register(serializer Serializer)

type Serializer 

type Serializer interface {
	Name() string
	ContentType() string
	Trusted() bool
	Marshal(any) ([]byte, error)
	Unmarshal([]byte, any) error
}

Serializer encodes and decodes queue payloads.

type ShutdownMode 

type ShutdownMode string
const (
	GracefulShutdown ShutdownMode = "graceful"
	WarmShutdown     ShutdownMode = "warm"
	ColdShutdown     ShutdownMode = "cold"
)

type Signature 

type Signature struct {
	Name    string
	Args    []any
	Kwargs  map[string]any
	Headers map[string]string
	Options SignatureOptions
}

Signature describes a task call before it becomes an envelope.

func NewSignature 

func NewSignature(name string, args ...any) Signature

func (Signature) Clone 

func (s Signature) Clone() Signature

func (Signature) WithCountdown 

func (s Signature) WithCountdown(countdown time.Duration, now ...time.Time) Signature

func (Signature) WithETA 

func (s Signature) WithETA(eta time.Time) Signature

func (Signature) WithExpires 

func (s Signature) WithExpires(expires time.Time) Signature

func (Signature) WithHeader 

func (s Signature) WithHeader(name string, value string) Signature

func (Signature) WithKwarg 

func (s Signature) WithKwarg(name string, value any) Signature

func (Signature) WithPriority 

func (s Signature) WithPriority(priority int) Signature

func (Signature) WithQueue 

func (s Signature) WithQueue(queue string) Signature

type SignatureOptions 

type SignatureOptions struct {
	Queue    string
	Priority int
	ETA      *time.Time
	Expires  *time.Time
}

SignatureOptions stores immutable task dispatch options.

type SolarEvent 

type SolarEvent string
const (
	SolarSunrise SolarEvent = "sunrise"
	SolarSunset  SolarEvent = "sunset"
)

type SolarProvider 

type SolarProvider interface {
	NextSolarEvent(SolarEvent, float64, float64, time.Time, *time.Location) (time.Time, bool)
}

type SolarSchedule 

type SolarSchedule struct {
	Event     SolarEvent
	Latitude  float64
	Longitude float64
	Location  *time.Location
	Provider  SolarProvider
}

func (SolarSchedule) Next 

func (s SolarSchedule) Next(after time.Time) (time.Time, bool)

func (SolarSchedule) NextRun 

func (s SolarSchedule) NextRun(lastRunAt *time.Time, now time.Time) (time.Time, bool)

type SoloPool 

type SoloPool struct{}

func NewSoloPool 

func NewSoloPool() *SoloPool

func (*SoloPool) Close 

func (p *SoloPool) Close(context.Context) error

func (*SoloPool) Run 

func (p *SoloPool) Run(ctx context.Context, executable PoolExecutable) (any, error)

func (*SoloPool) Strategy 

func (p *SoloPool) Strategy() PoolStrategy

type State 

type State string
const (
	StatePending  State = "PENDING"
	StateReceived State = "RECEIVED"
	StateStarted  State = "STARTED"
	StateRetry    State = "RETRY"
	StateSuccess  State = "SUCCESS"
	StateFailure  State = "FAILURE"
	StateRevoked  State = "REVOKED"
	StateIgnored  State = "IGNORED"
)

func (State) String 

func (s State) String() string

func (State) Terminal 

func (s State) Terminal() bool

type Task 

type Task struct {
	Name    string
	Func    TaskFunc
	Options TaskOptions
}

Task is one registered queue task.

type TaskDefinition 

type TaskDefinition struct {
	Name    string
	Func    TaskFunc
	Options TaskOptions
}

TaskDefinition is discovered from installed apps.

type TaskFunc 

type TaskFunc func(context.Context, ...any) (any, error)

TaskFunc is the executable task function contract.

type TaskOptions 

type TaskOptions struct {
	Serializer        string
	Queue             string
	RoutingKey        string
	Priority          int
	MaxRetries        int
	DefaultRetryDelay time.Duration
	RetryBackoff      bool
	RetryJitter       bool
	SoftTimeout       time.Duration
	HardTimeout       time.Duration
	RateLimit         RateLimit
	AckPolicy         AckPolicy
	IgnoreResult      bool
	TrackStarted      bool
}

TaskOptions contains Celery-style task execution options.

type TaskProvider 

type TaskProvider interface {
	QueueTasks() []TaskDefinition
}

TaskProvider is implemented by installed apps that expose queue tasks.

type Worker 

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

func NewWorker 

func NewWorker(app *App, broker Broker, backend ResultBackend, options WorkerOptions) *Worker

func (*Worker) ActiveTasks 

func (w *Worker) ActiveTasks() []ActiveTask

func (*Worker) Grow 

func (w *Worker) Grow(delta int)

func (*Worker) Heartbeat 

func (w *Worker) Heartbeat(ctx context.Context)

func (*Worker) MemoryLimitExceeded 

func (w *Worker) MemoryLimitExceeded() bool

func (*Worker) PrefetchLimit 

func (w *Worker) PrefetchLimit() int

func (*Worker) RestartPool 

func (w *Worker) RestartPool(ctx context.Context) error

func (*Worker) Run 

func (w *Worker) Run(ctx context.Context) error

func (*Worker) RunOnce 

func (w *Worker) RunOnce(ctx context.Context) error

func (*Worker) Shrink 

func (w *Worker) Shrink(delta int)

func (*Worker) Shutdown 

func (w *Worker) Shutdown(ctx context.Context, mode ShutdownMode) error

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context) error

func (*Worker) Stats 

func (w *Worker) Stats() WorkerStats

func (*Worker) TargetConcurrency 

func (w *Worker) TargetConcurrency(readyTasks int) int

type WorkerLogEntry 

type WorkerLogEntry struct {
	Event    string
	TaskID   string
	TaskName string
	Queue    string
	Hostname string
	State    State
	Error    string
	At       time.Time
	Fields   map[string]any
}

type WorkerLogger 

type WorkerLogger interface {
	LogWorkerEvent(context.Context, WorkerLogEntry)
}

type WorkerOptions 

type WorkerOptions struct {
	Hostname                string
	Queues                  []string
	Concurrency             int
	PrefetchMultiplier      int
	VisibilityTimeout       time.Duration
	PollInterval            time.Duration
	ShutdownTimeout         time.Duration
	AckPolicy               AckPolicy
	RejectOnWorkerLost      bool
	TrackStarted            bool
	MaxTasksPerWorkerChild  int
	MaxMemoryPerWorkerChild uint64
	Autoscale               AutoscaleConfig
	Pool                    Pool
	Logger                  WorkerLogger
	Events                  EventSink
	MemoryUsage             func() uint64
	Revocations             *RevocationRegistry
	RateLimiter             *RateLimiter
}

type WorkerStats 

type WorkerStats struct {
	Hostname                string
	Queues                  []string
	Concurrency             int
	PrefetchLimit           int
	PoolStrategy            PoolStrategy
	RejectOnWorkerLost      bool
	MaxTasksPerWorkerChild  int
	MaxMemoryPerWorkerChild uint64
	Processed               int
	Succeeded               int
	Failed                  int
	Revoked                 int
	RateLimited             int
	Acked                   int
	Nacked                  int
	Recycled                int
	Running                 int
}

Source Files

View all Source files

Directories

Path Synopsis
redis
Package redis provides a real Redis-backed queue result backend.
 Package redis provides a real Redis-backed queue result backend.
sql
rabbitmq
redis
Package redis provides a real Redis-backed queue broker for Gogo workers.
 Package redis provides a real Redis-backed queue broker for Gogo workers.
schedulers
redis
Package redis provides a real Redis-backed beat schedule store.
 Package redis provides a real Redis-backed beat schedule store.

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.