handler

Documentation

Index

• Constants
• Variables
• func Exec(ctx context.Context, handler Handler) (err error)
• func JobFromContext(ctx context.Context) (*jobs.Job, error)
• func WithContext(ctx context.Context, v CtxVars) context.Context
• type CtxVars
• type Func
• type Handler
  • func New(f Func, opts ...Option) (h Handler)
  • func (h *Handler) WithOptions(opts ...Option)
• type Option
  • func Concurrency(c int) Option
  • func Deadline(d time.Duration) Option
  • func MaxQueueCapacity(capacity int64) Option

Constants

const (
	DefaultHandlerDeadline = 30 * time.Second
)

Variables

var (
	CtxVarsKey           contextKey
	ErrContextHasNoJob   = errors.New("context has no Job")
	ErrNoHandlerForQueue = errors.New("no handler for queue")
	// TODO this error is here because cyclic imports with neoq
	ErrNoProcessorForQueue = errors.New("no processor configured for queue")
)

Functions

func Exec

func Exec(ctx context.Context, handler Handler) (err error)

Exec executes handler functions with a concrete time deadline

func JobFromContext

func JobFromContext(ctx context.Context) (*jobs.Job, error)

JobFromContext fetches the job from a context if the job context variable is already set

func WithContext

func WithContext(ctx context.Context, v CtxVars) context.Context

WithContext creates a new context with the job and transaction set

Types

type CtxVars

type CtxVars struct {
	Job *jobs.Job
	// this is a bit hacky. Tx here contains a pgx.Tx for PgBackend, but because we're in the handlers package, and we don't
	// want all neoq users to have pgx as a transitive dependency, we store Tx as any, and coerce it to a pgx.Tx inside
	// the postgres backend
	// TODO redesign HandlerCtxVars so it doesn't need to include a pgx.Tx
	Tx any
}

CtxVars are variables passed to every Handler context

type Func

type Func func(ctx context.Context) error

Func is a function that Handlers execute for every Job on a queue

type Handler

type Handler struct {
	Handle        Func
	Concurrency   int
	Deadline      time.Duration
	QueueCapacity int64
}

Handler handles jobs on a queue

func New

func New(f Func, opts ...Option) (h Handler)

New creates a new queue handler

func (*Handler) WithOptions

func (h *Handler) WithOptions(opts ...Option)

WithOptions sets one or more options on handler

type Option

type Option func(w *Handler)

Option is function that sets optional configuration for Handlers

func Concurrency

func Concurrency(c int) Option

Concurrency configures Neoq handlers to process jobs concurrently the default concurrency is the number of (v)CPUs on the machine running Neoq

func Deadline

func Deadline(d time.Duration) Option

Deadline configures handlers with a time deadline for every executed job The deadline is the amount of time that can be spent executing the handler's Func when a deadline is exceeded, the job is failed and enters its retry phase

func MaxQueueCapacity

func MaxQueueCapacity(capacity int64) Option

MaxQueueCapacity configures Handlers to enforce a maximum capacity on the queues that it handles queues that have reached capacity cause Enqueue() to block until the queue is below capacity