pipeline

type Batcher ¶

type Batcher[In, Batch any] interface {
	// Add adds an item. Returns true if the batch is ready to flush.
	Add(In) bool

	// Flush returns the current batch and resets. Returns false if empty.
	Flush() (Batch, bool)
}

Batcher accumulates input items and produces batches.

type ConfigurationOption ¶

type ConfigurationOption struct {
	// Default is the initial value of the configuration option.
	Default any
	// Name identifies the configuration option in facts.
	Name string
	// Description represents the help text about the configuration option.
	Description string
	// Flag corresponds to the CLI token with "--" prepended.
	Flag string
	// Type specifies the kind of the configuration option's value.
	Type ConfigurationOptionType
}

ConfigurationOption allows for the unified, retrospective way to setup PipelineItem-s.

func (opt ConfigurationOption) FormatDefault() string

type ConfigurationOptionType ¶

type ConfigurationOptionType int

ConfigurationOptionType represents the possible types of a ConfigurationOption's value.

const (
	// BoolConfigurationOption reflects the boolean value type.
	BoolConfigurationOption ConfigurationOptionType = iota
	// IntConfigurationOption reflects the integer value type.
	IntConfigurationOption
	// StringConfigurationOption reflects the string value type.
	StringConfigurationOption
	// FloatConfigurationOption reflects a floating point value type.
	FloatConfigurationOption
	// StringsConfigurationOption reflects the array of strings value type.
	StringsConfigurationOption
	// PathConfigurationOption reflects the file system path value type.
	PathConfigurationOption
)

func (opt ConfigurationOptionType) String() string

type DispatchFunc ¶

type DispatchFunc[Req any] func(ctx context.Context, req Req) error

DispatchFunc sends a request to a worker pool. The worker channel is captured in the closure, keeping the dispatch strategy decoupled from request semantics.

type Fetcher ¶

type Fetcher[Req, Resp any] interface {
	Fetch(ctx context.Context, req Req) (Resp, error)
}

Fetcher retrieves a response for a given request. It serves as the base interface for the cache decorator pattern: wrap a Fetcher with "check cache → fetch misses → update cache" logic.

type FetcherFunc ¶

type FetcherFunc[Req, Resp any] func(ctx context.Context, req Req) (Resp, error)

FetcherFunc adapts a plain function to the Fetcher interface.

func (f FetcherFunc[Req, Resp]) Fetch(ctx context.Context, req Req) (Resp, error)

type PassthroughBatcher ¶

type PassthroughBatcher[T any] struct {
	// contains filtered or unexported fields
}

PassthroughBatcher wraps each input item as a single-element batch. Add always returns true, meaning every item is immediately ready.

func (b *PassthroughBatcher[T]) Add(item T) bool

func (b *PassthroughBatcher[T]) Flush() ([]T, bool)

type Phase ¶

type Phase[S any] interface {
	Run(ctx context.Context, s S) (S, error)
}

Phase represents a single processing stage that transforms state S.

type PhaseFunc ¶

type PhaseFunc[S any] func(ctx context.Context, s S) (S, error)

PhaseFunc adapts a plain function to the Phase interface.

func (f PhaseFunc[S]) Run(ctx context.Context, s S) (S, error)

type RunPC ¶

type RunPC[In, Out, Job any] struct {
	// Buffer sets the capacity of the internal jobs channel.
	// Values below 1 are clamped to 1.
	Buffer int

	// Produce reads the input and sends work items on the jobs channel.
	Produce func(ctx context.Context, in In, jobs chan<- Job)

	// Consume reads work items from jobs and sends results on out.
	Consume func(ctx context.Context, jobs <-chan Job, out chan<- Out)
}

RunPC is a producer-consumer micro-skeleton that owns the goroutine topology: channel creation, goroutine spawning, and orderly shutdown.

Type parameters:

• In: the input consumed by the producer
• Out: the output emitted by the consumer
• Job: the internal work item flowing from producer to consumer

The Produce function reads from in and writes jobs to the jobs channel. The Consume function reads jobs and writes results to the out channel. Neither function should close its output channel; RunPC handles that.

func (r RunPC[In, Out, Job]) Run(ctx context.Context, in In) <-chan Out

type SharedResponse ¶

type SharedResponse[T any] struct {
	// contains filtered or unexported fields
}

SharedResponse evaluates a computation exactly once and caches the result for concurrent access by multiple goroutines. The computation receives a context.Context for cancellation support.

func NewSharedResponse[T any](compute func(context.Context) (T, error)) *SharedResponse[T]

func (s *SharedResponse[T]) Get(ctx context.Context) (T, error)

type ThresholdBatcher ¶

type ThresholdBatcher[T any] struct {
	// contains filtered or unexported fields
}

ThresholdBatcher accumulates items into a slice until the count reaches the configured threshold, at which point Add returns true.

func NewThresholdBatcher[T any](threshold int) *ThresholdBatcher[T]

func (b *ThresholdBatcher[T]) Add(item T) bool

func (b *ThresholdBatcher[T]) Flush() ([]T, bool)

type WorkerPool ¶

type WorkerPool[T any] struct {
	// MaxParallel is the maximum number of concurrent goroutines.
	// Zero defaults to runtime.NumCPU().
	MaxParallel int
	// Work processes a single item. Must not be nil.
	Work func(ctx context.Context, item T) error
}

WorkerPool runs Work on each item with at most MaxParallel goroutines. Returns the first non-nil error encountered, or nil. Remaining goroutines observe context cancellation on first error.

func (p WorkerPool[T]) Run(ctx context.Context, items []T) error

func (p WorkerPool[T]) RunChan(ctx context.Context, ch <-chan T) error