consumer

Documentation

Overview

Package consumer implements a job executor.

Index

• type Config
• type Consumer
  • func New(client client.Interface, runner Runner, providers ...Provider) *Consumer
  • func (c *Consumer) Run(ctx context.Context) error
  • func (c *Consumer) Stop()
• type Provider
  • func WithConfig(cfg Config) Provider
  • func WithLogger(logger logger.Logger) Provider
  • func WithQueue(queue string) Provider
• type Runner

Types

type Config

type Config struct {
	DequeueOpts     client.DequeueOpts `json:"dequeue_opts"`
	Concurrency     int                `json:"concurrency"`
	ShutdownTimeout time.Duration      `json:"shutdown_timeout"`
}

type Consumer

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

Consumer manages the dequeue, execution, and acknowledgement of jobs.

func New

func New(client client.Interface, runner Runner, providers ...Provider) *Consumer

New creates a new instance of Consumer.

func (*Consumer) Run

func (c *Consumer) Run(ctx context.Context) error

Run consumes jobs until Stop is called. After Stop is called, any currently running jobs will continue until completion.

func (*Consumer) Stop

func (c *Consumer) Stop()

Stop initiates the consumer's shutdown sequence.

type Provider

type Provider func(c *Consumer) *Consumer

Provider provides configuration to Consumers. It can be passed as an argument to New.

func WithConfig

func WithConfig(cfg Config) Provider

WithConfig allows custom configuration to be provided to new Consumers.

func WithLogger

func WithLogger(logger logger.Logger) Provider

WithLogger allows a custom logger to be provided to new Consumers.

func WithQueue

func WithQueue(queue string) Provider

WithQueue adds an additional target queue for new Consumers. If no queues are provided, all queues are dequeued from. Multiple queues can be consumed.

type Runner

type Runner interface {
	// Run executes a job. It can return a nil *JobResult, indicating success.
	// Returning an error will cause the job to fail.
	Run(ctx context.Context, job *resource.Job) (*resource.JobResult, error)
}

Runner can be implemented to execute jobs.