job

package
v0.0.0-...-e8da5e4 Latest Latest
Warning

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

 Go to latest Published: Aug 28, 2026 License: MIT Imports: 8 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// MaxMetadataValueBytes bounds every user-supplied metadata identity.
	MaxMetadataValueBytes = 256
	// MaxMetadataTags bounds user-supplied metadata dimensions.
	MaxMetadataTags = 32
	// MaxCorrelationFields bounds the transport-neutral correlation carrier.
	MaxCorrelationFields = 3
	// MaxCorrelationFieldBytes matches correlation codec field-name bounds.
	MaxCorrelationFieldBytes = 64
	// MaxCorrelationValueBytes matches the largest supported correlation ID.
	MaxCorrelationValueBytes = 1024
	// MaxTraceContextFields bounds caller-owned telemetry propagation carriers.
	MaxTraceContextFields = 16
	// MaxTraceContextFieldBytes bounds telemetry propagation field names.
	MaxTraceContextFieldBytes = 64
	// MaxTraceContextValueBytes bounds one telemetry propagation field value.
	MaxTraceContextValueBytes = 8192
	// MaxTraceContextBytes bounds the complete telemetry propagation carrier.
	MaxTraceContextBytes = 8192
)
View Source 
const DefaultMaxMessageBytes = 1024 * 1024

DefaultMaxMessageBytes bounds encoded messages to one mebibyte.

View Source 
const MaxRetryCount int64 = 100

MaxRetryCount bounds attempts scheduled from one delivered message.

Variables

View Source 
var (
	// ErrInvalidMessageLimit reports a non-positive decode limit.
	ErrInvalidMessageLimit = errors.New("message byte limit must be positive")
	// ErrMessageTooLarge reports an encoded message above its decode limit.
	ErrMessageTooLarge = errors.New("message exceeds byte limit")
	// ErrInvalidMessage reports unsafe or inconsistent execution metadata.
	ErrInvalidMessage = errors.New("invalid message execution metadata")
)

Functions

func Bool 

func Bool(val bool) *bool

Bool is a helper function that creates a pointer to a bool value. Useful for setting AllowOption fields that require *bool.

Example: 

opts := AllowOption{Jitter: job.Bool(true)}

func Encode 

func Encode(m *Message) []byte

Encode takes a Message struct and marshals it into a byte slice using msgpack. If the marshalling process encounters an error, the function will panic. It returns the marshalled byte slice.

Parameters:

  • m: A pointer to the Message struct to be encoded.

Returns:

  • A byte slice containing the msgpack-encoded data.

func Float64 

func Float64(val float64) *float64

Float64 is a helper function that creates a pointer to a float64 value. Useful for setting AllowOption fields that require *float64.

Example: 

opts := AllowOption{RetryFactor: job.Float64(1.5)}

func Int64 

func Int64(val int64) *int64

Int64 is a helper function that creates a pointer to an int64 value. Useful for setting AllowOption fields that require *int64.

Example: 

opts := AllowOption{RetryCount: job.Int64(3)}

func Time 

func Time(v time.Duration) *time.Duration

Time is a helper function that creates a pointer to a time.Duration value. Useful for setting AllowOption fields that require *time.Duration.

Example: 

opts := AllowOption{
    Timeout:    job.Time(5 * time.Minute),
    RetryDelay: job.Time(2 * time.Second),
}

Types

type AllowOption 

type AllowOption struct {
	RetryCount  *int64         // Maximum retry attempts (nil uses default: 0)
	RetryDelay  *time.Duration // Fixed delay between retries (nil uses exponential backoff)
	RetryFactor *float64       // Backoff multiplier (nil uses default: 2)
	RetryMin    *time.Duration // Minimum backoff delay (nil uses default: 100ms)
	RetryMax    *time.Duration // Maximum backoff delay (nil uses default: 10s)
	Jitter      *bool          // Enable backoff jitter (nil uses default: false)
	Timeout     *time.Duration // Job execution timeout (nil uses default: 60 minutes)
	Metadata    *Metadata      // Optional bounded identity propagated to operational records
}

AllowOption provides optional configuration for individual job execution. All fields are pointers to distinguish between "not set" and "set to zero value". This allows partial configuration while keeping unspecified fields at their defaults.

Example usage: 

opts := AllowOption{
    RetryCount: job.Int64(3),        // Retry failed jobs up to 3 times
    Timeout:    job.Time(5 * time.Minute), // 5 minute timeout
}
q.QueueTask(myTask, opts)

type Message 

type Message struct {
	Task TaskFunc `json:"-" msgpack:"-"`

	// Timeout is the duration the task can be processed by Handler.
	// zero if not specified
	// default is 60 time.Minute
	Timeout time.Duration `json:"timeout" msgpack:"timeout"`

	// Payload is the payload data of the task.
	Body []byte `json:"body" msgpack:"body"`

	// RetryCount set count of retry
	// default is 0, no retry.
	RetryCount int64 `json:"retry_count" msgpack:"retry_count"`

	// RetryDelay set delay between retry
	// default is 100ms
	RetryDelay time.Duration `json:"retry_delay" msgpack:"retry_delay"`

	// RetryFactor is the multiplying factor for each increment step.
	//
	// Defaults to 2.
	RetryFactor float64 `json:"retry_factor" msgpack:"retry_factor"`

	// Minimum value of the counter.
	//
	// Defaults to 100 milliseconds.
	RetryMin time.Duration `json:"retry_min" msgpack:"retry_min"`

	// Maximum value of the counter.
	//
	// Defaults to 10 seconds.
	RetryMax time.Duration `json:"retry_max" msgpack:"retry_max"`

	// Jitter eases contention by randomizing backoff steps
	Jitter bool `json:"jitter" msgpack:"jitter"`

	// Metadata is optional bounded identity retained for operational records.
	Metadata *Metadata `json:"metadata,omitempty" msgpack:"metadata,omitempty"`
	// contains filtered or unexported fields
}

Message describes a task and its metadata.

func Decode 

func Decode(b []byte) *Message

Decode takes a byte slice and unmarshals it into a Message struct using msgpack. If the unmarshalling process encounters an error, the function will panic. It returns a pointer to the unmarshalled Message.

Parameters:

  • b: A byte slice containing the msgpack-encoded data.

Returns:

  • A pointer to the decoded Message struct.

func DecodeE 

func DecodeE(b []byte, maxBytes int) (*Message, error)

DecodeE decodes a message while enforcing a positive encoded-message limit.

func NewMessage 

func NewMessage(m core.QueuedMessage, opts ...AllowOption) Message

NewMessage create new message

func NewTask 

func NewTask(task TaskFunc, opts ...AllowOption) Message

func (*Message) Ack 

func (m *Message) Ack() error

Ack acknowledges successful processing when the backend requires it.

func (*Message) AcknowledgementRequired 

func (m *Message) AcknowledgementRequired() bool

AcknowledgementRequired reports whether backend settlement is attached.

func (*Message) Bytes 

func (m *Message) Bytes() []byte

Bytes returns the byte slice of the Message struct. If the marshalling process encounters an error, the function will panic. It returns the marshalled byte slice.

Returns:

  • A byte slice containing the msgpack-encoded data.

func (*Message) CorrelationMetadata 

func (m *Message) CorrelationMetadata() map[string]string

CorrelationMetadata returns a copy of the transport-neutral carrier attached by a correlation-aware producer.

func (*Message) Nack 

func (m *Message) Nack() error

Nack rejects unsuccessful processing when the backend requires it.

func (*Message) NackFailure 

func (m *Message) NackFailure(err error) error

NackFailure rejects unsuccessful processing with its classified cause.

func (*Message) Payload 

func (m *Message) Payload() []byte

Payload returns the payload data of the Message. It returns the byte slice of the payload.

Returns:

  • A byte slice containing the payload data.

func (*Message) SetAcknowledgement 

func (m *Message) SetAcknowledgement(ack func() error, nack func() error)

SetAcknowledgement attaches backend delivery settlement callbacks.

func (*Message) SetFailureAcknowledgement 

func (m *Message) SetFailureAcknowledgement(
	ack func() error,
	nack func(error) error,
)

SetFailureAcknowledgement attaches settlement callbacks to a delivery whose backend needs the handler failure to choose retry or terminal behavior.

func (*Message) TraceContextMetadata 

func (m *Message) TraceContextMetadata() map[string]string

TraceContextMetadata returns a copy of the transport-neutral telemetry carrier attached by a trace-aware producer.

func (*Message) Validate 

func (m *Message) Validate() error

Validate checks that message execution metadata is finite and bounded.

type Metadata 

type Metadata struct {
	OriginalID           string            `json:"original_id,omitempty" msgpack:"original_id,omitempty"`
	PayloadSchemaVersion string            `json:"payload_schema_version,omitempty" msgpack:"payload_schema_version,omitempty"`
	ContentType          string            `json:"content_type,omitempty" msgpack:"content_type,omitempty"`
	EnqueuedAt           *time.Time        `json:"enqueued_at,omitempty" msgpack:"enqueued_at,omitempty"`
	RetryPolicy          string            `json:"retry_policy,omitempty" msgpack:"retry_policy,omitempty"`
	HandlerType          string            `json:"handler_type,omitempty" msgpack:"handler_type,omitempty"`
	JobType              string            `json:"job_type,omitempty" msgpack:"job_type,omitempty"`
	Tags                 map[string]string `json:"tags,omitempty" msgpack:"tags,omitempty"`
	TraceID              string            `json:"trace_id,omitempty" msgpack:"trace_id,omitempty"`
	TenantID             string            `json:"tenant_id,omitempty" msgpack:"tenant_id,omitempty"`
	ProducerVersion      string            `json:"producer_version,omitempty" msgpack:"producer_version,omitempty"`
	Correlation          map[string]string `json:"correlation,omitempty" msgpack:"correlation,omitempty"`
	TraceContext         map[string]string `json:"trace_context,omitempty" msgpack:"trace_context,omitempty"`
}

Metadata carries optional, backend-neutral job identity into failure and dead-letter records. Empty fields remain unknown and must not be fabricated.

func (Metadata) Validate 

func (m Metadata) Validate() error

Validate rejects unbounded identity, tag, and time metadata.

type Options 

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

Options configures retry behavior and timeouts for individual jobs. These settings control how failed tasks are retried and when they time out.

func NewOptions 

func NewOptions(opts ...AllowOption) Options

NewOptions creates a job Options struct by merging defaults with provided AllowOption values. Only non-nil fields in AllowOption will override the defaults. This allows partial configuration where unspecified options retain their default values.

Example: 

// Only override retry count and timeout, keep other defaults
opts := job.NewOptions(job.AllowOption{
    RetryCount: job.Int64(5),
    Timeout:    job.Time(10 * time.Minute),
})

type TaskFunc 

type TaskFunc func(context.Context) error

TaskFunc is the task function

Source Files

View all Source files

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.