beanstalkjobs

Documentation

Index

• type Configurer
• type ConnPool
  • func NewConnPool(network, address, tName string, tout time.Duration, log *zap.Logger) (*ConnPool, error)
  • func (cp *ConnPool) Delete(_ context.Context, id uint64) error
  • func (cp *ConnPool) Put(_ context.Context, body []byte, pri uint32, delay, ttr time.Duration) (uint64, error)
  • func (cp *ConnPool) Reserve(reserveTimeout time.Duration) (uint64, []byte, error)
  • func (cp *ConnPool) Stats(_ context.Context) (map[string]string, error)
  • func (cp *ConnPool) Stop()
• type Driver
  • func FromConfig(tracer *sdktrace.TracerProvider, configKey string, log *zap.Logger, ...) (*Driver, error)
  • func FromPipeline(tracer *sdktrace.TracerProvider, pipe jobs.Pipeline, log *zap.Logger, ...) (*Driver, error)
  • func (d *Driver) Pause(ctx context.Context, p string) error
  • func (d *Driver) Push(ctx context.Context, jb jobs.Message) error
  • func (d *Driver) Resume(ctx context.Context, p string) error
  • func (d *Driver) Run(ctx context.Context, p jobs.Pipeline) error
  • func (d *Driver) State(ctx context.Context) (*jobs.State, error)
  • func (d *Driver) Stop(ctx context.Context) error
• type Item
  • func (i *Item) Ack() error
  • func (i *Item) Body() []byte
  • func (i *Item) Context() ([]byte, error)
  • func (i *Item) GroupID() string
  • func (i *Item) Headers() map[string][]string
  • func (i *Item) ID() string
  • func (i *Item) Nack() error
  • func (i *Item) NackWithOptions(redeliver bool, delay int) error
  • func (i *Item) Priority() int64
  • func (i *Item) Requeue(headers map[string][]string, delay int) error
  • func (i *Item) Respond(_ []byte, _ string) error
• type Options
  • func (o *Options) DelayDuration() time.Duration

Types

type Configurer

type Configurer interface {
	// UnmarshalKey takes a single key and unmarshals it into a Struct.
	UnmarshalKey(name string, out any) error
	// Has checks if a config section exists.
	Has(name string) bool
}

type ConnPool

type ConnPool struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewConnPool

func NewConnPool(network, address, tName string, tout time.Duration, log *zap.Logger) (*ConnPool, error)

func (*ConnPool) Delete

func (cp *ConnPool) Delete(_ context.Context, id uint64) error

func (*ConnPool) Put

func (cp *ConnPool) Put(_ context.Context, body []byte, pri uint32, delay, ttr time.Duration) (uint64, error)

Put the payload TODO use the context ??

func (*ConnPool) Reserve

func (cp *ConnPool) Reserve(reserveTimeout time.Duration) (uint64, []byte, error)

Reserve reserves and returns a job from one of the tubes in t. If no job is available before time timeout has passed, Reserve returns a ConnError recording ErrTimeout.

Typically, a client will reserve a job, perform some work, then delete the job with Conn.Delete.

func (*ConnPool) Stats

func (cp *ConnPool) Stats(_ context.Context) (map[string]string, error)

func (*ConnPool) Stop

func (cp *ConnPool) Stop()

Stop and close the connections

type Driver

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

func FromConfig

func FromConfig(tracer *sdktrace.TracerProvider, configKey string, log *zap.Logger, cfg Configurer, pipe jobs.Pipeline, pq jobs.Queue) (*Driver, error)

func FromPipeline

func FromPipeline(tracer *sdktrace.TracerProvider, pipe jobs.Pipeline, log *zap.Logger, cfg Configurer, pq jobs.Queue) (*Driver, error)

func (*Driver) Pause

func (d *Driver) Pause(ctx context.Context, p string) error

func (*Driver) Push

func (d *Driver) Push(ctx context.Context, jb jobs.Message) error

func (*Driver) Resume

func (d *Driver) Resume(ctx context.Context, p string) error

func (*Driver) Run

func (d *Driver) Run(ctx context.Context, p jobs.Pipeline) error

func (*Driver) State

func (d *Driver) State(ctx context.Context) (*jobs.State, error)

State https://github.com/beanstalkd/beanstalkd/blob/master/doc/protocol.txt#L514

func (*Driver) Stop

func (d *Driver) Stop(ctx context.Context) error

type Item

type Item struct {
	// Job contains the pluginName of job broker (usually PHP class).
	Job string `json:"job"`
	// Ident is a 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 []byte `json:"payload"`

	// Options contain a set of PipelineOptions specific to job execution. Can be empty.
	Options *Options `json:"options,omitempty"`
	// contains filtered or unexported fields
}

func (*Item) Ack

func (i *Item) Ack() error

func (*Item) Body

func (i *Item) Body() []byte

Body packs job payload into binary payload.

func (*Item) Context

func (i *Item) Context() ([]byte, error)

Context packs job context (job, id) into binary payload. Not used in the sqs, MessageAttributes used instead

func (*Item) GroupID

func (i *Item) GroupID() string

func (*Item) Headers

func (i *Item) Headers() map[string][]string

func (*Item) ID

func (i *Item) ID() string

func (*Item) Nack

func (i *Item) Nack() error

func (*Item) NackWithOptions

func (i *Item) NackWithOptions(redeliver bool, delay int) error

func (*Item) Priority

func (i *Item) Priority() int64

func (*Item) Requeue

func (i *Item) Requeue(headers map[string][]string, delay int) error

func (*Item) Respond

func (i *Item) Respond(_ []byte, _ string) error

type Options

type Options struct {
	// Priority is job priority, default - 10
	// pointer to distinguish 0 as a priority and nil as a 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 int `json:"delay,omitempty"`
	// AutoAck option
	AutoAck bool `json:"auto_ack"`
	// Beanstalk Tube
	Queue string `json:"queue,omitempty"`
	// contains filtered or unexported fields
}

Options carry information about how to handle a given job.

func (*Options) DelayDuration

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

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