pipeline

package module
v2.1.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 10, 2026 License: MIT Imports: 8 Imported by: 0

README

pipeline

A Go package for building concurrent data processing pipelines using channels.

Overview

pipeline provides a set of composable stages for processing data streams concurrently. It handles context cancellation, error propagation, and goroutine lifecycle management automatically.

All pipeline stages require a name parameter as their first argument. These names are used to create trace regions for performance analysis using Go's runtime/trace package.

Installation

go get github.com/schraf/pipeline

Usage

Basic Example

To create a pipeline, you define a PipelineConfig which specifies the pipeline's name, buffer sizes, and an Executor function. The Executor is where you connect your pipeline stages.

package main

import (
	"context"
	"fmt"

	"github.com/schraf/pipeline"
)

func main() {
	// Define the pipeline configuration
	cfg := pipeline.PipelineConfig[int, int]{
		Name:            "example",
		InputChannels:   1,
		OutputChannels:  1,
		Executor: func(pipe *pipeline.Pipe, in pipeline.MultiChannelReceiver[int], out pipeline.MultiChannelSender[int]) {
			// Connect input to output through a Transform stage
			pipeline.Transform("multiply", pipe, func(ctx context.Context, x int) (*int, error) {
				result := x * 2
				return &result, nil
			}, in.At(0), out.At(0))
		},
	}

	// Create the pipeline
	p, _ := pipeline.NewPipeline(context.Background(), cfg)

	// Start processing
	p.Start()

	// Feed data into the pipeline
	go func() {
		defer p.CloseAllInputs()
		for i := 1; i <= 10; i++ {
			p.Inputs().At(0) <- i
		}
	}()

	// Consume results
	for v := range p.Outputs().At(0) {
		fmt.Println(v)
	}

	// Wait for completion and check for errors
	if err := p.Wait(); err != nil {
		panic(err)
	}
}
Chaining Pipelines

You can connect a pipeline to one or more downstream pipelines using the Chain function. This connects the outputs of the previous pipeline to the inputs of the new pipelines automatically. If multiple configurations are provided, the data is fanned out to all downstream pipelines.

// Create first pipeline (e.g., generates or transforms data)
p1, _ := pipeline.NewPipeline(ctx, cfg1)

// Chain two pipelines to the first one.
// The output of p1 will be broadcast to both p2 and p3.
// Note: Chain uses p1's context for the new pipelines.
downstreamPipelines, err := pipeline.Chain(p1, cfg2, cfg3)
if err != nil {
    panic(err)
}

// Start all pipelines
p1.Start()
for _, p := range downstreamPipelines {
    p.Start()
}

// Feed p1 and consume from p2/p3 (via downstreamPipelines)
// ...
Pipeline Groups

The PipelineGroup struct allows you to manage a collection of pipelines with the same input and output types. It provides methods to add, start, and wait for multiple pipelines as a single unit.

// Create a group for the downstream pipelines
group := pipeline.NewPipelineGroup[int, int]()

// Add pipelines to the group
group.Add(downstreamPipelines...)

// Start all pipelines in the group
if err := group.Start(); err != nil {
    panic(err)
}

// Wait for all pipelines in the group to complete
if err := group.Wait(ctx); err != nil {
    panic(err)
}

Pipeline Stages

Stages are designed to be used inside the Executor function of your PipelineConfig. The pipe argument provided to the executor is passed to each stage.

Transform

Applies a transformation function to each value:

pipeline.Transform("transform", pipe, func(ctx context.Context, x int) (*int, error) {
    result := x * 2
    return &result, nil
}, in, out)
Filter

Filters values based on a predicate:

pipeline.Filter("filter", pipe, func(ctx context.Context, x int) (bool, error) {
    return x%2 == 0, nil
}, in, out)
Batch

Groups values into fixed-size batches:

pipeline.Batch("batch", pipe, func(ctx context.Context, batch []int) (*int, error) {
    sum := 0
    for _, v := range batch {
        sum += v
    }
    return &sum, nil
}, 3, in, out)
ParallelTransform

Applies transformation with concurrent workers:

pipeline.ParallelTransform("parallel-transform", pipe, 5, func(ctx context.Context, x int) (*int, error) {
    result := x * 2
    return &result, nil
}, in, out)
FanIn

Merges multiple input channels into one:

pipeline.FanIn("fan-in", pipe, out, in1, in2, in3)
FanOut

Distributes values to multiple output channels (broadcast):

pipeline.FanOut("fan-out", pipe, in, out1, out2, out3)
FanOutRoundRobin

Distributes values round-robin style:

pipeline.FanOutRoundRobin("fan-out-round-robin", pipe, in, out1, out2, out3)
Limit

Limits the number of values passed through:

pipeline.Limit("limit", pipe, 10, in, out)
Split

Routes values to different channels based on a selector:

pipeline.Split("split", pipe, func(ctx context.Context, x int) int {
    return (x - 1) % 3
}, in, out1, out2, out3)
Aggregate

Collects all values into a single slice:

pipeline.Aggregate("aggregate", pipe, in, out)
Reduce

Processes values incrementally using a reducer function, combining them with an accumulator. This allows aggregating results as they come in without keeping all values in memory:

pipeline.Reduce("reduce", pipe, 0, func(ctx context.Context, acc int, x int) (int, error) {
    return acc + x, nil
}, in, out)
Flatten

Takes an input channel of slices and emits each element of each slice as an individual item on the output channel:

pipeline.Flatten("flatten", pipe, in, out)
Expand

Takes single input items from a channel and for each input, outputs multiple items of another type. The expander function returns an iterator (iter.Seq2[Out, error]) of output items for each input, allowing for lazy evaluation and avoiding loading all expanded items into memory at once:

pipeline.Expand("expand", pipe, func(ctx context.Context, x int) iter.Seq2[string, error] {
    return func(yield func(string, error) bool) {
        yield(fmt.Sprintf("%d", x), nil)
        yield(fmt.Sprintf("%d", x*2), nil)
    }
}, in, out)

Error Handling

The pipeline automatically cancels all stages when an error occurs. The first error encountered is returned by Wait():

if err := p.Wait(); err != nil {
    log.Fatal(err)
}

Requirements

  • Go 1.24.0 or later

License

See LICENSE file for details.

Documentation

Overview

Package pipeline provides composable stages for building concurrent data processing pipelines using channels. It handles context cancellation, error propagation, and goroutine lifecycle management automatically.

Index

Constants

View Source 
const (
	StateInvalid int32 = iota
	StateCreated
	StateStarted
	StateWaiting
	StateDone
)

Variables

This section is empty.

Functions

func Aggregate 

func Aggregate[T any](name string, pipe *Pipe, in <-chan T, out chan<- []T)

Aggregate consumes all values from the input channel and sends the collected slice of values as a single item on the output channel.

func Batch 

func Batch[In any, Out any](name string, pipe *Pipe, batcher func(context.Context, []In) (*Out, error), batchSize int, in <-chan In, out chan<- Out)

Batch groups incoming values into fixed-size batches, passes each batch to the batcher function, and forwards the resulting value to the output channel. Any remaining items after the input channel closes are processed as a final batch. The batcher must return a non-nil pointer when err is nil, otherwise a panic will occur.

func Expand 

func Expand[In any, Out any](name string, pipe *Pipe, expander func(context.Context, In) iter.Seq2[Out, error], in <-chan In, out chan<- Out)

Expand reads values from the input channel, applies the expander function to each value, and forwards all items from the returned iterator to the output channel. For each input item, the expander returns an iterator of output items, which are all sent to the output channel. Processing continues until the context is done or the input channel is closed. This allows for lazy evaluation and avoids loading all expanded items into memory at once.

func ExpandSlice 

func ExpandSlice[T any](name string, pipe *Pipe, in <-chan []T, out chan<- T)

ExpandSlice reads slices from the input channel and forwards all items from the returned iterator to the output channel.

func FanIn 

func FanIn[T any](name string, pipe *Pipe, out chan<- T, in ...<-chan T)

FanIn merges multiple input channels into a single output channel, forwarding all values from each input until the context is done or all inputs are closed.

func FanOut 

func FanOut[T any](name string, pipe *Pipe, in <-chan T, out ...chan<- T)

FanOut distributes items from a single input channel to multiple output channels, sending each item to all output channels.

func FanOutRoundRobin 

func FanOutRoundRobin[T any](name string, pipe *Pipe, in <-chan T, out ...chan<- T)

FanOutRoundRobin distributes items from a single input channel to multiple output channels using round-robin distribution, sending each item to only one output channel. Panics if no output channels are provided.

func Filter 

func Filter[T any](name string, pipe *Pipe, filter func(context.Context, T) (bool, error), in <-chan T, out chan<- T)

Filter reads values from the input channel, applies the filter predicate, and forwards only values that satisfy the predicate to the output channel. It respects context cancellation and stops processing on error.

func Flatten 

func Flatten[T any](name string, pipe *Pipe, in <-chan []T, out chan<- T)

Flatten takes an input channel of slices and emits each element of each slice as an individual item on the output channel. It continues until the input channel is closed or the context is cancelled.

func Limit 

func Limit[T any](name string, pipe *Pipe, n int, in <-chan T, out chan<- T)

Limit reads values from the input channel, forwards at most n values to the output channel, and then returns. It respects context cancellation while reading and forwarding values.

func ParallelTransform 

func ParallelTransform[In any, Out any](name string, pipe *Pipe, workers int, transformer func(context.Context, In) (*Out, error), in <-chan In, out chan<- Out)

ParallelTransform applies the transformer function to values read from the input channel using a fixed number of concurrent workers, forwarding successful results to the output channel until the context is done or the input channel is closed. The transformer must return a non-nil pointer when err is nil, otherwise a panic will occur.

func Reduce 

func Reduce[T any, Acc any](name string, pipe *Pipe, initial Acc, reducer func(context.Context, Acc, T) (Acc, error), in <-chan T, out chan<- Acc)

Reduce processes values from the input channel incrementally using a reducer function, combining them with an accumulator. This allows aggregating results as they come in without keeping all values in memory. The reducer function takes the current accumulator and the next value, and returns the updated accumulator. The final accumulated result is sent to the output channel.

func Split 

func Split[T any](name string, pipe *Pipe, selector func(context.Context, T) int, in <-chan T, out ...chan<- T)

Split routes each value read from the input channel to exactly one of the provided output channels, as determined by the selector function. The selector must return a valid index into the out slice. Panics if the selector returns an invalid index.

func Transform 

func Transform[In any, Out any](name string, pipe *Pipe, transformer func(context.Context, In) (*Out, error), in <-chan In, out chan<- Out)

Transform reads values from the input channel, applies the transformer function, and forwards successful results to the output channel until the context is done or the input channel is closed. The transformer must return a non-nil pointer when err is nil, otherwise a panic will occur.

Types

type MultiChannelReceiver 

type MultiChannelReceiver[T any] []chan T

func (MultiChannelReceiver[T]) At 

func (m MultiChannelReceiver[T]) At(index int) <-chan T

func (MultiChannelReceiver[T]) Iter 

func (m MultiChannelReceiver[T]) Iter() iter.Seq[<-chan T]

func (MultiChannelReceiver[T]) Len 

func (m MultiChannelReceiver[T]) Len() int

type MultiChannelSender 

type MultiChannelSender[T any] []chan T

func (MultiChannelSender[T]) At 

func (m MultiChannelSender[T]) At(index int) chan<- T

func (MultiChannelSender[T]) Iter 

func (m MultiChannelSender[T]) Iter() iter.Seq[chan<- T]

func (MultiChannelSender[T]) Len 

func (m MultiChannelSender[T]) Len() int

func (MultiChannelSender[T]) Send 

func (m MultiChannelSender[T]) Send(ctx context.Context, index int, values ...T) error

type Pipe 

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

Pipe is used in defining stages in a PipelineExecutor function.

func (Pipe) Context added in v2.1.2 

func (p Pipe) Context() context.Context

Context returns the context of the pipeline

type Pipeline 

type Pipeline[In any, Out any] struct {
	// contains filtered or unexported fields
}

Pipeline coordinates concurrent processing stages, managing their lifecycle and propagating errors and cancellation signals across all stages.

func Chain 

func Chain[In any, Middle any, Out any](
	prev *Pipeline[In, Middle],
	configs ...PipelineConfig[Middle, Out],
) ([]*Pipeline[Middle, Out], error)

Chain connects the outputs of the previous pipeline to the inputs of new pipelines created from the provided configurations. It manages the data transfer between the pipelines automatically.

The connection handles data transfer: - Output/input counts must match for a 1-to-1 connection. - Outputs will fan-out to each new pipeline

Data transfer stops and the new pipelines' inputs are closed when the previous pipeline's outputs are closed.

func NewPipeline 

func NewPipeline[In any, Out any](ctx context.Context, cfg PipelineConfig[In, Out]) (*Pipeline[In, Out], context.Context)

NewPipeline creates a new Pipeline and a derived context for coordinating pipeline stages. The returned context is cancelled when any stage encounters an error. Use the returned Pipeline to register stages and wait for completion.

func (*Pipeline[In, Out]) CloseAllInputs 

func (p *Pipeline[In, Out]) CloseAllInputs()

CloseAllInputs will close all of the input channels

func (*Pipeline[In, Out]) Context 

func (p *Pipeline[In, Out]) Context() context.Context

func (*Pipeline[In, Out]) Inputs 

func (p *Pipeline[In, Out]) Inputs() MultiChannelSender[In]

func (*Pipeline[In, Out]) Outputs 

func (p *Pipeline[In, Out]) Outputs() MultiChannelReceiver[Out]

func (*Pipeline[In, Out]) Start 

func (p *Pipeline[In, Out]) Start() error

func (*Pipeline[In, Out]) State 

func (p *Pipeline[In, Out]) State() *PipelineState

func (*Pipeline[In, Out]) Wait 

func (p *Pipeline[In, Out]) Wait() error

Wait blocks until all registered stages complete and returns the first error encountered by any stage, or nil if all stages completed successfully.

type PipelineConfig 

type PipelineConfig[In any, Out any] struct {
	Name             string
	InputChannels    int
	InputBufferSize  int
	OutputChannels   int
	OutputBufferSize int
	StartImmediately bool
	Executor         PipelineExecutor[In, Out]
}

PipeConfig defines the make up of a pipeline and is required for construction of it

type PipelineExecutor 

type PipelineExecutor[In any, Out any] func(*Pipe, MultiChannelReceiver[In], MultiChannelSender[Out])

PipeExecutor defines the body of the pipeline. The function should connect the input channel to the output channel using stages on the provided pipe.

type PipelineGroup 

type PipelineGroup[In any, Out any] struct {
	// contains filtered or unexported fields
}

func NewPipelineGroup 

func NewPipelineGroup[In any, Out any]() *PipelineGroup[In, Out]

func (*PipelineGroup[In, Out]) Add 

func (g *PipelineGroup[In, Out]) Add(pipelines ...*Pipeline[In, Out])

func (*PipelineGroup[In, Out]) Start 

func (g *PipelineGroup[In, Out]) Start() error

func (*PipelineGroup[In, Out]) Wait 

func (g *PipelineGroup[In, Out]) Wait(ctx context.Context) error

type PipelineState 

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

func (*PipelineState) Is 

func (s *PipelineState) Is(state int32) bool

func (*PipelineState) String 

func (s *PipelineState) String() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.