worker

Documentation

Index

• Variables
• func GetWorkerID(ctx log.ValueContext) (string, bool)
• func GetWorkerName(ctx log.ValueContext) (string, bool)
• func NewWorkerContext(parentCtx context.Context, name string, id string) context.Context
• func ProvideAsLongRunningWorker(provider any, anns ...fx.Annotation) fx.Option
• type ConcurrencyConfig
• type ConcurrencyMode
  • func (i ConcurrencyMode) String() string
• type Handler
• type LongRunningWorker
• type LongRunningWorkerHandler
  • func NewLongRunningWorkerHandler(conf config.Config, lf *log.LoggerFactory, db database.DB, ...) *LongRunningWorkerHandler
  • func NewLongRunningWorkerHandlerFx(params LongRunningWorkerHandlerParams) *LongRunningWorkerHandler
  • func (h *LongRunningWorkerHandler) Run() error
  • func (h *LongRunningWorkerHandler) Start()
  • func (h *LongRunningWorkerHandler) Stop(ctx context.Context)
• type LongRunningWorkerHandlerParams

Variables

var ErrCodeWorkerError = "WORKER_ERROR"

var ErrWorkerHandlerNoWorkers = errors.Newf("WORKER_HANDLER_NO_WORKERS", "worker handler has no workers")

var ModuleWorkerHandler = di.FxProvideAs[Handler](
	NewLongRunningWorkerHandlerFx,
	nil,
	[]fx.Annotation{fx.ResultTags(`group:"worker_handlers"`)},
)

Functions

func GetWorkerID

func GetWorkerID(ctx log.ValueContext) (string, bool)

func GetWorkerName

func GetWorkerName(ctx log.ValueContext) (string, bool)

func NewWorkerContext

func NewWorkerContext(
	parentCtx context.Context,
	name string,
	id string,
) context.Context

NewWorkerContext creates a new child context with worker info and a configured logger with the worker name and id. Dependencies should be added to the context using NewContextWithDeps.

func ProvideAsLongRunningWorker

func ProvideAsLongRunningWorker(provider any, anns ...fx.Annotation) fx.Option

Types

type ConcurrencyConfig

type ConcurrencyConfig struct {
	Mode          ConcurrencyMode
	SingleLockTTL time.Duration
}

type ConcurrencyMode

type ConcurrencyMode int

const (
	ConcurrencyModeMulti  ConcurrencyMode = iota
	ConcurrencyModeSingle ConcurrencyMode = iota
)

func (ConcurrencyMode) String

func (i ConcurrencyMode) String() string

type Handler

type Handler interface {
}

type LongRunningWorker

type LongRunningWorker interface {
	GetName() string
	GetID() string
	GetConcurrency() ConcurrencyConfig
	Run(ctx context.Context) error
}

type LongRunningWorkerHandler

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

func NewLongRunningWorkerHandler

func NewLongRunningWorkerHandler(
	conf config.Config,
	lf *log.LoggerFactory,
	db database.DB,
	fxLifecycle fx.Lifecycle,
	fxShutdowner fx.Shutdowner,
	dlFactory distributedlock.Factory,
	workers []LongRunningWorker,
) *LongRunningWorkerHandler

func NewLongRunningWorkerHandlerFx

func NewLongRunningWorkerHandlerFx(params LongRunningWorkerHandlerParams) *LongRunningWorkerHandler

func (*LongRunningWorkerHandler) Run

func (h *LongRunningWorkerHandler) Run() error

Run runs all workers asynchronously and blocks until all workers are done. It returns error if any of the workers returns error from LongRunningWorker.Run

func (*LongRunningWorkerHandler) Start

func (h *LongRunningWorkerHandler) Start()

Start starts all workers asynchronously and returns immediately

func (*LongRunningWorkerHandler) Stop

func (h *LongRunningWorkerHandler) Stop(ctx context.Context)

type LongRunningWorkerHandlerParams

type LongRunningWorkerHandlerParams struct {
	di.BaseParams
	DLFactory distributedlock.Factory
	Workers   []LongRunningWorker `group:"long_running_workers"`
}