jobs

Documentation

Index

• Constants
• type Acknowledger
• type Constructor
• type Consumer
• type Job
• type Options
  • func (o *Options) DelayDuration() time.Duration
• type State

Constants

const (
	RRID       string = "rr_id"
	RRJob      string = "rr_job"
	RRHeaders  string = "rr_headers"
	RRPipeline string = "rr_pipeline"
	RRDelay    string = "rr_delay"
	RRPriority string = "rr_priority"
	RRAutoAck  string = "rr_auto_ack"
)

constant keys to pack/unpack messages from different drivers

Types

type Acknowledger

type Acknowledger interface {
	// Ack - acknowledge the Item after processing
	Ack() error

	// Nack - discard the Item
	Nack() error

	// Requeue - put the message back to the queue with the optional delay
	Requeue(headers map[string][]string, delay int64) error

	// Respond to the queue
	Respond(payload []byte, queue string) error
}

Acknowledger provides queue specific item management

type Constructor

type Constructor interface {
	ConsumerFromConfig(configKey string, queue pq.Queue) (Consumer, error)
	ConsumerFromPipeline(pipe *pipeline.Pipeline, queue pq.Queue) (Consumer, error)
}

Constructor constructs Consumer interface. Endure abstraction.

type Consumer

type Consumer interface {
	Push(ctx context.Context, job *Job) error
	Register(ctx context.Context, pipeline *pipeline.Pipeline) error
	Run(ctx context.Context, pipeline *pipeline.Pipeline) error
	Stop(ctx context.Context) error

	Pause(ctx context.Context, pipeline string)
	Resume(ctx context.Context, pipeline string)

	// State provide information about driver state
	State(ctx context.Context) (*State, error)
}

Consumer represents a single jobs driver interface

type Job

type Job struct {
	// Job contains name of job broker (usually PHP class).
	Job string `json:"job"`

	// Ident is unique identifier of the job, should be provided from outside
	Ident string `json:"id"`

	// Payload is string data (usually JSON) passed to Job broker.
	Payload string `json:"payload"`

	// Headers with key-value pairs
	Headers map[string][]string `json:"headers"`

	// Options contains set of PipelineOptions specific to job execution. Can be empty.
	Options *Options `json:"options,omitempty"`
}

Job carries information about single job.

type Options

type Options struct {
	// Priority is job priority, default - 10
	// pointer to distinguish 0 as a priority and nil as priority not set
	Priority int64 `json:"priority"`

	// Pipeline manually specified pipeline.
	Pipeline string `json:"pipeline,omitempty"`

	// Delay defines time duration to delay execution for. Defaults to none.
	Delay int64 `json:"delay,omitempty"`

	// AutoAck use to ack a job right after it arrived from the driver
	AutoAck bool `json:"auto_ack"`

	// kafka specific fields
	// Topic is kafka topic
	Topic string `json:"topic"`
	// Optional metadata
	Metadata string `json:"metadata"`
	// Kafka partition
	Partition int32 `json:"partition"`
	// Kafka offset
	Offset int64 `json:"offset"`
}

Options carry information about how to handle given job.

func (*Options) DelayDuration

func (o *Options) DelayDuration() time.Duration

DelayDuration returns delay duration in a form of time.Duration.

type State

type State struct {
	// Pipeline name
	Pipeline string
	// Driver name
	Driver string
	// Queue name (tube for the beanstalk)
	Queue string
	// Active jobs which are consumed from the driver but not handled by the PHP worker yet
	Active int64
	// Delayed jobs
	Delayed int64
	// Reserved jobs which are in the driver but not consumed yet
	Reserved int64
	// Status - 1 Ready, 0 - Paused
	Ready bool
	// New in 2.10.5, pipeline priority
	Priority uint64
}

State represents job's state