pipeline

Documentation

Index

• func IsProcessorSkipped(err error) bool
• type ErrProcessorSkipped
  • func (e *ErrProcessorSkipped) Error() string
• type Hooks
• type Job
• type Metrics
• type NoopMetrics
  • func (n NoopMetrics) RecordJob(duration time.Duration, succeeded bool)
  • func (n NoopMetrics) SetInFlightJobs(count int64, workerCount int)
  • func (n NoopMetrics) SetPendingJobs(count int64)
• type Pipeline
  • func NewPipeline[T any, R any](stages []*Stage[T, R], accumulator func(T, R), merger func(R, R) R, ...) *Pipeline[T, R]
  • func (p *Pipeline[T, R]) Run(ctx context.Context, pctx T) error
• type Processor
• type Stage
  • func NewConcurrentStage[T any, R any](name string, number int, processors ...Processor[T, R]) *Stage[T, R]
  • func NewStage[T any, R any](name string, number int, processors ...Processor[T, R]) *Stage[T, R]
• type WorkerPool
  • func NewWorkerPool[J Job](config WorkerPoolConfig, ...) *WorkerPool[J]
  • func (p *WorkerPool[J]) Start(ctx context.Context) error
  • func (p *WorkerPool[J]) Stop()
  • func (p *WorkerPool[J]) Wait()
• type WorkerPoolConfig

Functions

func IsProcessorSkipped

func IsProcessorSkipped(err error) bool

IsProcessorSkipped returns true if err is ErrProcessorSkipped.

Types

type ErrProcessorSkipped

type ErrProcessorSkipped struct {
	Reason string
}

ErrProcessorSkipped signals that a processor was skipped.

func (*ErrProcessorSkipped) Error

func (e *ErrProcessorSkipped) Error() string

type Hooks

type Hooks[T any, R any] interface {
	OnStageStart(ctx context.Context, pctx T, stage *Stage[T, R]) error
	OnStageEnd(ctx context.Context, stage *Stage[T, R], pctx T, mergedResult R, err error) error
	OnProcessorStart(ctx context.Context, pctx T, stage *Stage[T, R], proc Processor[T, R])
	OnProcessorEnd(ctx context.Context, pctx T, stage *Stage[T, R], proc Processor[T, R], result R, err error, duration time.Duration)
}

Hooks allows project-specific logic to be injected into the pipeline execution.

type Job

type Job interface {
	ID() any
}

Job represents a unit of work.

type Metrics

type Metrics interface {
	// RecordJob records the completion of a single job.
	RecordJob(duration time.Duration, succeeded bool)

	// SetPendingJobs updates the count of jobs waiting in the database.
	SetPendingJobs(count int64)

	// SetInFlightJobs updates the count of jobs currently being processed.
	SetInFlightJobs(count int64, workerCount int)
}

Metrics defines the interface for pipeline performance tracking.

type NoopMetrics

type NoopMetrics struct{}

NoopMetrics provides a neutral implementation of the Metrics interface.

func (NoopMetrics) RecordJob

func (n NoopMetrics) RecordJob(duration time.Duration, succeeded bool)

func (NoopMetrics) SetInFlightJobs

func (n NoopMetrics) SetInFlightJobs(count int64, workerCount int)

func (NoopMetrics) SetPendingJobs

func (n NoopMetrics) SetPendingJobs(count int64)

type Pipeline

type Pipeline[T any, R any] struct {
	// contains filtered or unexported fields
}

Pipeline coordinates the execution of stages.

func NewPipeline

func NewPipeline[T any, R any](
	stages []*Stage[T, R],
	accumulator func(T, R),
	merger func(R, R) R,
	hooks Hooks[T, R],
	metrics Metrics,
	logger *slog.Logger,
) *Pipeline[T, R]

NewPipeline creates a new Pipeline.

func (*Pipeline[T, R]) Run

func (p *Pipeline[T, R]) Run(ctx context.Context, pctx T) error

Run executes the full pipeline for a given context.

type Processor

type Processor[T any, R any] interface {
	Name() string
	Process(ctx context.Context, pctx T) (R, error)
}

Processor is the interface for a single unit of work in the pipeline. T is the context type, R is the result type.

type Stage

type Stage[T any, R any] struct {
	Name       string
	Number     int
	Processors []Processor[T, R]
	Concurrent bool // Whether processors in this stage can run concurrently
}

Stage groups related processors into a logical pipeline stage.

func NewConcurrentStage

func NewConcurrentStage[T any, R any](name string, number int, processors ...Processor[T, R]) *Stage[T, R]

NewConcurrentStage creates a new concurrent stage.

func NewStage

func NewStage[T any, R any](name string, number int, processors ...Processor[T, R]) *Stage[T, R]

NewStage creates a new sequential stage.

type WorkerPool

type WorkerPool[J Job] struct {
	// contains filtered or unexported fields
}

WorkerPool manages concurrent job processing.

func NewWorkerPool

func NewWorkerPool[J Job](
	config WorkerPoolConfig,
	fetchFn func(context.Context, int) ([]J, int64, error),
	processFn func(context.Context, J) error,
	metrics Metrics,
	logger *slog.Logger,
) *WorkerPool[J]

NewWorkerPool creates a new production-ready worker pool.

func (*WorkerPool[J]) Start

func (p *WorkerPool[J]) Start(ctx context.Context) error

Start begins the worker pool and polls for jobs.

func (*WorkerPool[J]) Stop

func (p *WorkerPool[J]) Stop()

Stop initiates a graceful shutdown.

func (*WorkerPool[J]) Wait

func (p *WorkerPool[J]) Wait()

Wait blocks until the pool has fully stopped.

type WorkerPoolConfig

type WorkerPoolConfig struct {
	WorkerCount         int           // Number of concurrent workers (default: 1)
	PollInterval        time.Duration // Interval between job polls (default: 5s)
	ProgressLogInterval time.Duration // Interval between progress logs (default: 30s, 0 to disable)
}

WorkerPoolConfig configures the worker pool.