consumer

Documentation

Index

• type Config
• type Consumer
  • func New[P any, S any, T Processor[P, S]](cfg Config[P, S, T]) (*Consumer[P, S, T], error)
  • func (c *Consumer[P, S, T]) Start(ctx context.Context) (err error)
• type Processor

Types

type Config

type Config[P any, S any, T Processor[P, S]] struct {

	// Workers represent the maximum number of Jobs that can be in-flight at one time by having a
	// maximum number of workers.
	// Default will be set to one.
	Workers int

	// Pollers indicate the maximum number of polling workers trying to retrieve Jobs for processing.
	// this should only be tuned for consistent high-flow services for which a single poller becomes the
	// bottleneck which is rare. This should never be set greater than the maximum number of workers and in
	// most cases should be far less than.
	// By default, there will only be one poller which should be all you need 99.99999% of the time.
	Pollers int

	// EnableAutocomplete turns on auto-completion of a Job that is processed without error.
	EnableAutocomplete bool

	// Processor is the main processor of Jobs.
	Processor T

	// Queue is the Jbo Queue for which to pull jobs from for processing.
	Queue string

	// Client represents the pre-configured low-level relay client.
	Client *relay.Client[P, S]

	// Backoff if the backoff used when calling the `next` or `complete` endpoint and there is no data yet
	// available.
	// Optional: If not set a default backoff is used.
	Backoff backoff.Exponential
}

Config contains all configuration data to initialize a Consumer for use.

type Consumer

type Consumer[P any, S any, T Processor[P, S]] struct {
	// contains filtered or unexported fields
}

Consumer is a wrapper around the low-level Relay Client to abstract away polling and distribution of Jobs for processing.

func New

func New[P any, S any, T Processor[P, S]](cfg Config[P, S, T]) (*Consumer[P, S, T], error)

New initializes a new Consumer for use.

func (*Consumer[P, S, T]) Start

func (c *Consumer[P, S, T]) Start(ctx context.Context) (err error)

Start initializes the workers and starts polling for new jobs.

type Processor

type Processor[P any, S any] interface {
	// Process processes the Job.
	//
	// NOTE: An error returned by this will NOT bubble up beyond this function, it's the responsibility of each
	//       Processor to handle its own errors. This signature only exists to allow wrapping the Processor like
	//       middleware to intercept it and perform actions based upon it.
	Process(context.Context, *relay.JobHelper[P, S]) error
}

Processor represents a processor of Jobs