pipeline

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. When creating a pipeline with WithPipeline, you also provide a name for the overall pipeline task.

Installation

go get github.com/schraf/pipeline

Usage

Basic Example

package main

import (
	"context"
	"fmt"
	"slices"

	"github.com/schraf/pipeline"
)

func main() {
	p, _ := pipeline.WithPipeline(context.Background(), "example")

	// Define some data to process
	data := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

	// Create channels
	in := make(chan int, len(data))
	out := make(chan int, len(data))

	// Source: feed data into the pipeline from a slice
	pipeline.SourceSlice("source", p, slices.Values(data), in)

	// Transform: multiply by 2
	pipeline.Transform("multiply", p, func(ctx context.Context, x int) (*int, error) {
		result := x * 2
		return &result, nil
	}, in, out)


	// Read results
	for v, err := range pipeline.Sink("sink", p, out) {
        if err != nil {
            panic(err)
        }

		fmt.Println(v)
	}
}

Pipeline Stages

Source

Starts a pipeline from a fallible iterator (iter.Seq2[T, error]). If the iterator returns an error, the pipeline is cancelled.

// Example: create a custom iterator that reads from a file or database
// and can return an error.
var myIterator iter.Seq2[string, error] 

out := make(chan string, 100)
pipeline.Source("my-source", p, myIterator, out)

SourceSlice

Starts a pipeline from a simple, error-free iterator (iter.Seq[T]), which is useful for in-memory slices.

data := []int{1, 2, 3, 4, 5}
out := make(chan int, 5)

// The iterator can be created from a slice using slices.Values
pipeline.SourceSlice("source", p, slices.Values(data), out)

Sink

The "opposite" of a source, Sink takes an input channel and returns an iter.Seq2[T, error] iterator. This allows consuming values from a pipeline using a standard for...range loop. If the pipeline is cancelled, the iterator will yield an error.

// Given 'out' is a channel from a pipeline stage...
it := pipeline.Sink("sink", p, out)

// You can now iterate over the results.
for v, err := range it {
    if err != nil {
        // This can happen if the pipeline is cancelled.
        log.Fatalf("iterator error: %v", err)
    }
    fmt.Println(v)
}

Transform

Applies a transformation function to each value:

pipeline.Transform("transform", p, 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", p, 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", p, 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", p, 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", p, out, in1, in2, in3)

FanOut

Distributes values to multiple output channels (broadcast):

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

FanOutRoundRobin

Distributes values round-robin style:

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

Limit

Limits the number of values passed through:

pipeline.Limit("limit", p, 10, in, out)

Split

Routes values to different channels based on a selector:

pipeline.Split("split", p, 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", p, 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", p, 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", p, 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", p, 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.