pipefn

README

pipefn

pipefn is a Go library for building lazy, functional-style streaming pipelines on top of iter.Seq.

status

This project is still a work in progress. Expect bugs and possible API changes as it evolves.

Use in production at your own risk.

Documentation

Overview

Package pipefn provides functional-style and composable transformations for data streams, enabling streaming pipelines without intermediate buffering.

This package is built around the concept of Pipes, a Pipe[T] represents a lazily-evaluated stream of values of type T paired with an internal error channel.

All transformations (Map, FlatMap, Filter, and more) are provided as package-level functions. Each transformation returns a new Pipe, allowing pipelines to be composed through simple chaining. Values are only produced when the resulting stream is iterated, making all pipelines demand-driven.

Errors produced by any stage flow through the pipe’s internal error channel. All transformations inherit their input Pipe’s error channel, so errors automatically propagate through the pipeline and can be consumed alongside values when the pipeline is processed.

Pipes are consumed using the Results, Values, ForEach, Collect and CollectValues methods. Consuming a Pipe is a destructive operation, once a Pipe has been consumed, it cannot be reused or iterated again.

For more details on each type, function, and available transformation, please refer to the corresponding package-level documentation.

Example

Example demonstrates a relatively complex pipeline that parses items from multiple sources

package main

import (
	"fmt"
	"iter"
	"strings"
	"sync"

	"github.com/KasperOmsK/pipefn"
)

type Item struct {
	ID string
}

type Event struct {
	Type  string
	Items []Item
}

// Example demonstrates a relatively complex pipeline that parses items from multiple sources
func main() {

	var (
		// stream is a fake stream that will yield some data but then report
		// a terminal failure.
		stream pipefn.Stream[string] = newStubStream([]string{
			"purchase:9999",
		}, fmt.Errorf("fake error"))
	)

	// Create a pipe from a slice
	inputsFromSlice := pipefn.FromSlice([]string{
		"purchase:1001,1002,1003",
		"refund:2001",
		"purchase:1004,1005",
		"purchase:", // invalid item (empty ID)
	})

	// ... or from an iter.Seq
	inputsFromSeq := pipefn.FromSeq(func(yield func(string) bool) {
		yield("refund:420")
	})

	// ... or from a channel
	ch := make(chan string)
	go func() {
		ch <- ":1000" // invalid item (no event type)
		ch <- "refund:1,2,3"
		close(ch)
	}()
	inputsFromChan, _ := pipefn.FromChan(ch)

	// ... or from the Stream interface
	inputsFromStream := pipefn.From(stream)

	// Merge multiple pipes of the same type.
	//
	// if any input pipe fails, the merged pipe will also fail with the first encountered failure.
	allInputs := pipefn.Merge(inputsFromSeq, inputsFromSlice, inputsFromStream, inputsFromChan)

	// Map applies deterministic transformations that cannot fail.
	// Transformations are simple functions, making it easy to reuse existing code.
	trimmed := pipefn.Map(allInputs, strings.TrimSpace)

	// TryMap applies transformations that may fail.
	// Errors are forwarded to the Pipe's error channel.
	events := pipefn.TryMap(trimmed, func(line string) (Event, error) {
		return ParseEvent(line)
	})

	// All common FP-style transformation are available.

	purchases := pipefn.Filter(events, func(e Event) bool {
		return e.Type == "purchase"
	})

	items := pipefn.FlatMap(purchases, func(e Event) []Item {
		return e.Items
	})

	// TryMap can act as a validation filter with error reporting
	validItems := pipefn.TryMap(items, func(it Item) (Item, error) {
		if it.ID == "" {
			return Item{}, fmt.Errorf("missing item ID")
		}
		return it, nil
	})

	batches := pipefn.Chunk(validItems, 2)

	// retreive the value stream and the pipeline errors channel
	values, errs := batches.Results()

	// Pipeline errors *must* be consumed concurrently to avoid blocking the pipeline.
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for err := range errs {
			// errors emitted errs will always be of type *PipeError
			perr := err.(*pipefn.PipeError)
			fmt.Printf("pipeline error: item: %v, reason: %s\n", perr.Item, perr.Reason)
		}
	}()

	// Or alternatively, if you're only interested in successful values of the pipeline...
	//	values := batches.Values()
	// ... will only return the value stream and silently discard pipeline errors.

	batchCount := 0
	for batch := range values.Seq() {
		fmt.Println("batch:", batchCount)
		err := ProcessBatch(batch)
		if err != nil {
			break // Stopping consumption halts the pipeline
		}
		batchCount++
	}
	wg.Wait()

	// values.Err() returns any terminal failure of the pipe.
	//
	// A non-nil error here indicates that at least one input stream failed.
	//
	// In our example, let's say that such an error means the work done by ProcessBatch
	// should be discarded,
	if err := values.Err(); err != nil {
		Rollback()
	} else {
		Commit()
	}
}

func Rollback() {
	fmt.Println("work rollbacked :()")
}

func Commit() {
	fmt.Println("work commited !")
}

func ParseEvent(line string) (Event, error) {
	parts := strings.SplitN(line, ":", 2)
	if len(parts) != 2 {
		return Event{}, fmt.Errorf("invalid event format: %q", line)
	}

	eventType := strings.TrimSpace(parts[0])
	rawItems := strings.TrimSpace(parts[1])

	var items []Item
	if rawItems == "" {
		return Event{}, fmt.Errorf("invalid event format: no items")
	}

	for _, id := range strings.Split(rawItems, ",") {
		items = append(items, Item{
			ID: strings.TrimSpace(id),
		})
	}

	return Event{
		Type:  eventType,
		Items: items,
	}, nil
}

func ProcessBatch(items []Item) error {
	for _, it := range items {
		fmt.Println(it)
	}
	return nil
}

// newStubStream returns a pipefn.Stream that simulates reading from a file with a terminal failure
func newStubStream(data []string, err error) pipefn.Stream[string] {
	return &stubStream{
		stubData: func(yield func(string) bool) {
			for _, e := range data {
				if !yield(e) {
					return
				}
			}
		},
		failure: err,
	}
}

type stubStream struct {
	stubData iter.Seq[string]
	failure  error
}

func (fs *stubStream) Seq() iter.Seq[string] {
	return fs.stubData
}

func (fs *stubStream) Err() error {
	return fs.failure
}

Index

• Variables
• type MapFunc
• type Pipe
  • func Chunk[T any](p Pipe[T], chunkSize int) Pipe[[]T]
  • func Concat[T any](pipes ...Pipe[T]) Pipe[T]
  • func Drop[T any](p Pipe[T], n int) Pipe[T]
  • func Filter[T any](p Pipe[T], predicate Predicate[T]) Pipe[T]
  • func FlatMap[In, Out any](p Pipe[In], fn MapFunc[In, []Out]) Pipe[Out]
  • func FlatTryMap[In, Out any](p Pipe[In], fn TryMapFunc[In, []Out]) Pipe[Out]
  • func Flatten[T any](p Pipe[[]T]) Pipe[T]
  • func From[T any](s Stream[T]) Pipe[T]
  • func FromChan[T any](ch <-chan T) (p Pipe[T], done <-chan struct{})
  • func FromSeq[T any](seq iter.Seq[T]) Pipe[T]
  • func FromSlice[T any](elems []T) Pipe[T]
  • func GroupByKey[T any, K comparable](p Pipe[T], keyFunc func(T) K) Pipe[[]T]
  • func GroupByKeyAggregate[In any, K comparable, Out any](p Pipe[In], keyFunc func(In) K, initFunc func(first In) Out, ...) Pipe[Out]
  • func Map[In, Out any](p Pipe[In], fn MapFunc[In, Out]) Pipe[Out]
  • func Merge[T any](pipes ...Pipe[T]) Pipe[T]
  • func Take[T any](p Pipe[T], n int) Pipe[T]
  • func TryMap[In, Out any](p Pipe[In], fn TryMapFunc[In, Out]) Pipe[Out]
  • func Zip[Left, Right any](left Pipe[Left], right Pipe[Right]) Pipe[Tuple[Left, Right]]
  • func (p *Pipe[T]) Collect() ([]T, []error, error)
  • func (p *Pipe[T]) CollectValues() ([]T, error)
  • func (p *Pipe[T]) ForEach(consumeFn func(item T), errorFn func(err error)) error
  • func (p *Pipe[T]) Results() (values Stream[T], errs <-chan error)
  • func (p *Pipe[T]) Tap(tapFn func(T))
  • func (p *Pipe[T]) Values() Stream[T]
• type PipeError
  • func (pe *PipeError) Error() string
  • func (pe *PipeError) Unwrap() error
• type Predicate
• type Stream
• type TryMapFunc
• type Tuple

Examples

• Package
• Chunk
• Concat
• Drop
• Filter
• FlatMap
• FlatTryMap
• Flatten
• FromChan
• GroupByKey
• GroupByKeyAggregate
• Map
• Merge
• Take
• TryMap
• Zip

Variables

var (
	ErrAlreadyConsumed = errors.New("pipefn: pipe has already been consumed")
)

Types

type MapFunc

type MapFunc[In, Out any] func(in In) Out

MapFunc is a pure mapping function used by Map that transforms a value of type In into a value of type Out.

type Pipe

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

Pipe represents a lazily-evaluated sequence of values of type T.

A Pipe is created using one of the From* functions (e.g. From, FromSlice etc.) and transformed into new Pipes using transformation functions such as Map, TryMap, Filter, GroupBy etc. Each transformation produces a new Pipe that wraps the previous one, forming a pipeline.

Evaluation is lazy: values flow through the pipeline only when iterating over the output.

A Pipe is consumed by calling a consumption function such as:

• Results
• Values
• ForEach
• Collect
• CollectValues

Consumption is global and occurs at most once. After the first call to any of the above function, the entire pipeline is considered consumed, even if iteration as not begun. Any subsequent attempt to call any consumption function on any Pipe derived from that pipeline returns an empty stream whose Err() method returns ErrAlreadyConsumed.

Errors are handled at two levels:

• Per-item errors produced by fallible stages (e.g. TryMap) are sent on the pipeline's error channel and are associated with individual items.

  Such errors *must* be handled upon final iteration, either by manually draining the error channel (see Results) or by explicitely discarding them (see Values).

• A pipeline may also terminate with a terminal error, which is reported by the value stream's Err() method once consumption completes.

  Terminal errors represent failures of the underlying data source or execution environment (e.g. I/O errors, network failures, database interruptions), rather than issues tied to specific items flowing through the pipeline.

  This package does not define or interpret such errors. Instead, it leaves the responsability to define these errors to the user through the Stream interface. In particular, when a Pipe is created via From, its terminal error is exactly the one returned by the provided Stream. See Stream and From for details on how terminal errors are produced and observed.

The zero Pipe is an empty pipe that yields no values and no errors.

func Chunk

func Chunk[T any](p Pipe[T], chunkSize int) Pipe[[]T]

Chunk groups incoming values into slices of the given size and returns a Pipe producing those slices.

The final chunk may be smaller than chunkSize.

If chunkSize is zero or negative, all input values are emitted as a single chunk.

Transformation errors and terminal errors from the input pipe are propagated unchanged.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	flat := pipefn.FromSlice([]int{1, 2, 3, 4, 5})

	chunked := pipefn.Chunk(flat, 2)

	values, _, _ := chunked.Collect()

	for _, v := range values {
		fmt.Println(v)
	}
}

Output:
[1 2]
[3 4]
[5]

func Concat

func Concat[T any](pipes ...Pipe[T]) Pipe[T]

Concat combines multiple pipes into a single Pipe that emits all values and pipeline errors produced by the input pipes, processing the pipes sequentially in the order they are provided.

Values and errors produced by earlier pipes are emitted before values and errors from later pipes.

Once iteration begins, all input pipes are considered used. If a pipe terminates early with an error, all subsequent pipes are still marked as used. Iterating over them will produce a Stream whose Err() returns ErrAlreadyConsumed.

If pipes is empty, Concat returns an empty pipe.

terminal errors are reported through the concatenated value stream. After the stream has been fully consumed, Stream.Err() returns the first terminal error encountered among the input streams, if any. If a pipe terminates with a failure, subsequent pipes are not processed.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	p1 := pipefn.FromSlice([]int{1, 2})
	p2 := pipefn.FromSlice([]int{3})
	p3 := pipefn.FromSlice([]int{4, 5})

	concat := pipefn.Concat(p1, p2, p3)

	values, _, _ := concat.Collect()

	// Print concatenated values
	for _, v := range values {
		fmt.Println(v)
	}

}

Output:
1
2
3
4
5

func Drop

func Drop[T any](p Pipe[T], n int) Pipe[T]

Drop returns a Pipe that skips the first n elements from the input Pipe p and produces all subsequent elements.

If n is zero or negative, the returned Pipe yields all elements from p. If p contains fewer than n elements, the returned Pipe produces no values.

All transformation errors from the input Pipe are propagated unchanged, including those associated with skipped elements.

The returned Pipe preserves the terminal error of the input Pipe.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	input := pipefn.FromSlice([]int{1, 2, 3, 4, 5})
	take := pipefn.Drop(input, 3)
	values, _, _ := take.Collect()

	for _, v := range values {
		fmt.Println(v)
	}
}

Output:
4
5

func Filter

func Filter[T any](p Pipe[T], predicate Predicate[T]) Pipe[T]

Filter returns a Pipe that yields only the values for which predicate returns true.

Transformation errors and terminal errors from the input pipe are propagated unchanged.

Filter panics if predicate is nil.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	input := pipefn.FromSlice([]int{1, 2, 3, 4, 5})

	// Keep only even numbers
	filtered := pipefn.Filter(input, func(n int) bool {
		return n%2 == 0
	})

	values, _, _ := filtered.Collect()
	for _, v := range values {
		fmt.Println(v)
	}

}

Output:
2
4

func FlatMap

func FlatMap[In, Out any](p Pipe[In], fn MapFunc[In, []Out]) Pipe[Out]

FlatMap transforms each input value using fn and returns a Pipe producing the flattened output values.

FlatMap is equivalent to calling Flatten(Map(p, fn)).

Transformation errors and terminal errors from the input pipe are propagated unchanged.

FlatMap panics if fn is nil.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	input := pipefn.FromSlice([]int{1, 2, 3})

	// For each number, produce a slice of the number and multiple of 10
	flatMapped := pipefn.FlatMap(input, func(in int) []int {
		return []int{in, in * 10}
	})

	values, _, _ := flatMapped.Collect()
	for _, v := range values {
		fmt.Println(v)
	}
}

Output:
1
10
2
20
3
30

func FlatTryMap

func FlatTryMap[In, Out any](p Pipe[In], fn TryMapFunc[In, []Out]) Pipe[Out]

FlatTryMap applies fn to each value produced by p and returns a new Pipe emitting the flattened successful results.

If fn returns a non-nil error, the error is wrapped in a *PipeError and sent to the Pipe's error channel, and the corresponding value is skipped.

Each successful result slice returned by fn is flattened, and its elements are emitted individually by the returned Pipe.

Transformation errors produced by FlatTryMap are added to the input pipe's error channel. Any terminal error from the input pipe is propagated unchanged.

FlatTryMap is equivalent to calling Flatten(TryMap(p, fn)).

FlatTryMap panics if fn is nil.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	input := pipefn.FromSlice([]int{1, -2, 3})

	// For each number, produce a slice of the number and multiple of 10 if positive
	flatTryMapped := pipefn.FlatTryMap(input, func(in int) ([]int, error) {
		if in < 0 {
			return nil, fmt.Errorf("negative number")
		}
		return []int{in, in * 10}, nil
	})

	values, errs, _ := flatTryMapped.Collect()

	for _, v := range values {
		fmt.Println(v)
	}

	for _, e := range errs {
		perr := e.(*pipefn.PipeError)
		fmt.Printf("error: item=%v, reason=%s\n", perr.Item, perr.Reason)
	}

}

Output:
1
10
3
30
error: item=-2, reason=negative number

func Flatten

func Flatten[T any](p Pipe[[]T]) Pipe[T]

Flatten converts a Pipe of slices into a Pipe of their elements, emitting the items of each slice in order.

Transformation errors and terminal errors from the input pipe are propagated unchanged.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	groups := pipefn.FromSlice([][]int{{1, 2}, {3, 4}})

	flattened := pipefn.Flatten(groups)
	values, _, _ := flattened.Collect()

	for _, v := range values {
		fmt.Println(v)
	}
}

Output:
1
2
3
4

func From

func From[T any](s Stream[T]) Pipe[T]

From creates a pipe that produces values from the provided stream.

The returned pipe will have the same terminal error as s, if any

func FromChan

func FromChan[T any](ch <-chan T) (p Pipe[T], done <-chan struct{})

FromChan returns a Pipe that produces values received from ch, along with a done channel that is closed when the pipe stops consuming from ch.

The resulting Pipe yields each value read from the channel until the channel is closed. If the downstream consumer stops iteration early, iteration stops immediately and remaining values in the channel are not consumed.

FromChan does not close the provided channel; instead, the done channel is closed when consumption of ch terminates, either because the channel is fully drained or because iteration stops early. Callers producing values into ch should monitor done to know when to stop sending.

The returned Pipe never terminates with an error: its value stream's Err() method always returns nil.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	ch := make(chan int)

	p, done := pipefn.FromChan(ch)

	// Producer
	go func() {
		defer close(ch)
		for i := 0; i < 10; i++ {
			select {
			case ch <- i:
			case <-done:
				fmt.Println("producer: stopping early")
				return
			}
		}

	}()

	i := 0
	// Consumer: only take first 3 values
	for v := range p.Values().Seq() {
		fmt.Println(v)
		i++
		if i >= 3 {
			break
		}
	}
	// wait for producer to stop to ensure that the output
	// is consistent
	<-ch

}

Output:
0
1
2
producer: stopping early

func FromSeq

func FromSeq[T any](seq iter.Seq[T]) Pipe[T]

FromSeq creates a Pipe that produces values from the provided iter.Seq.

The returned Pipe never terminates with an error: its value stream's Err() method always returns nil.

FromSeq panics if seq is nil.

func FromSlice

func FromSlice[T any](elems []T) Pipe[T]

FromSlice creates a Pipe that produces the elements of elems in order.

The resulting Pipe yields each element of the slice sequentially. The slice is not copied; iteration reads directly from elems.

If elems is nil, the resulting Pipe behaves as an empty pipe.

The returned Pipe never terminates with an error: its value stream's Err() method always returns nil.

func GroupByKey

func GroupByKey[T any, K comparable](p Pipe[T], keyFunc func(T) K) Pipe[[]T]

GroupByKey groups consecutive input values according to a key function and returns a Pipe producing slices of those grouped values.

GroupByKey does not reorder values; it relies on the input Pipe already being ordered by the grouping key if consistent grouping is desired.

In other words, Values are grouped only when they appear consecutively with the same key. When the key returned by keyFunc changes, the current group is emitted and a new group is started.

For example, given input values:

A, A, B, B, A

GroupByKey will emit:

[A, A], [B, B], [A]

Transformation errors and terminal errors from the input pipe are propagated unchanged.

GroupByKey panics if keyFunc is nil.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {

	input := pipefn.FromSlice([]string{"apple", "apricot", "banana", "blueberry", "cherry", "avocado"})

	// Group strings by their first letter
	grouped := pipefn.GroupByKey(input, func(s string) string {
		return string(s[0])
	})

	values, _, _ := grouped.Collect()
	for _, group := range values {
		fmt.Println(group)
	}

	// Note that since the input is not preordered by grouping key, "avocado" appears after "cherry"

}

Output:
[apple apricot]
[banana blueberry]
[cherry]
[avocado]

func GroupByKeyAggregate

func GroupByKeyAggregate[In any, K comparable, Out any](
	p Pipe[In],
	keyFunc func(In) K,
	initFunc func(first In) Out,
	updateFunc func(acc *Out, item In)) Pipe[Out]

GroupByKeyAggregate groups input values by key and aggregates them using user-supplied initialization and update callbacks, producing one aggregated output value per group.

GroupByKeyAggregate is equivalent to performing a GroupByKey followed by a Map, but does so without allocating a slice for each group. This makes it preferred for pipelines where groups may be large.

initFunc is called when a new group starts. It receives the first value of the group and should returns the initial accumulator for that group.

updateFunc is called for each value in the current group, including the first one. It receives a pointer to the accumulator and the current input value, and should updates the accumulator in place.

For example, to sum values in each group:

initFunc := func(v int) int {
    return 0 // start at 0
}

updateFunc := func(acc *int, v int) {
    *acc += v // add the value to the accumulator
}

Like GroupBy, GroupByKeyAggregate does not reorder input values. The input Pipe must already be ordered by key if consistent aggregation per key is desired.

For example, with input values:

A1, A2, B1, B2, A3

GroupByKeyAggregate will emit aggregated results for:

[A1, A2], [B1, B2], [A3]

Transformation errors and terminal errors from the input pipe are propagated unchanged.

GroupByKeyAggregate panics if either keyFunc, initFunc or updateFunc is nil.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	input := pipefn.FromSlice(
		[]int{
			1, 2,
			10, 10, 10,
			20, 25,
			5,
		})

	// Group by tens digit and sum values within each consecutive group
	keyFunc := func(v int) int {
		return v / 10 // key = 10s digit
	}
	initFunc := func(_ int) int {
		return 0 // start sum at 0
	}
	updateFunc := func(acc *int, v int) {
		*acc += v // add the value to the accumulator
	}

	grouped := pipefn.GroupByKeyAggregate(input, keyFunc, initFunc, updateFunc)

	values, _, _ := grouped.Collect()
	for _, v := range values {
		fmt.Println(v)
	}

	// Note that since the input is not preordered by grouping key, the last single digit value appears at the end

}

Output:
3
30
45
5

func Map

func Map[In, Out any](p Pipe[In], fn MapFunc[In, Out]) Pipe[Out]

Map returns a Pipe that applies fn to each value produced by p.

Transformation errors and terminal errors from the input pipe are propagated unchanged.

Map panics if fn is nil.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	input := pipefn.FromSlice([]int{1, 2, 3})

	squared := pipefn.Map(input, func(in int) int { return in * in })

	values, _, _ := squared.Collect()
	for _, v := range values {
		fmt.Println(v)
	}
}

Output:
1
4
9

func Merge

func Merge[T any](pipes ...Pipe[T]) Pipe[T]

Merge combines multiple pipes into a single Pipe that emits all values and pipeline errors produced by the input pipes.

Values and errors originating from different pipes may be interleaved and are not guaranteed to preserve the relative ordering between pipes.

If pipes is empty, Merge returns an empty pipe.

Terminal errors are reported through the merged value stream. After the stream has been fully consumed, Stream.Err() returns the first terminal error encountered among the merged pipes, if any.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	p1 := pipefn.FromSlice([]int{1, 2})
	p2 := pipefn.FromSlice([]int{3})
	p3 := pipefn.FromSlice([]int{4, 5})

	merged := pipefn.Merge(p1, p2, p3)

	values, _, _ := merged.Collect()

	for _, v := range values {
		fmt.Println(v)
	}

	// Possible output:
	// 1
	// 3
	// 2
	// 4
	// 5
}

func Take

func Take[T any](p Pipe[T], n int) Pipe[T]

Take returns a Pipe that produces at most n elements from p.

Iteration stops after n values have been yielded, or when p is exhausted, whichever comes first. If n is zero or negative, the returned Pipe produces no values.

Because the returned Pipe stops consuming p after n elements, any transformation errors that would occur beyond the first n elements are never produced.

The returned Pipe preserves the terminal error of the input Pipe.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	input := pipefn.FromSlice([]int{1, 2, 3, 4, 5})
	take := pipefn.Take(input, 3)
	values, _, _ := take.Collect()

	for _, v := range values {
		fmt.Println(v)
	}
}

Output:
1
2
3

func TryMap

func TryMap[In, Out any](p Pipe[In], fn TryMapFunc[In, Out]) Pipe[Out]

TryMap applies fn to each value produced by p and returns a new Pipe emitting the successful results.

If fn returns a non-nil error, the error is wrapped in a *PipeError and sent to the Pipe's error channel, and the corresponding value is skipped.

Transformation errors produced by TryMap are added to the input pipe's error channel. Any terminal error from the input pipe is propagated unchanged.

TryMap panics if fn is nil.

Example

package main

import (
	"fmt"
	"math"

	"github.com/KasperOmsK/pipefn"
)

func main() {

	input := pipefn.FromSlice([]int{1, -2, 16})

	squared := pipefn.TryMap(input, func(in int) (float64, error) {
		if in < 0 {
			return 0, fmt.Errorf("NaN")
		}
		return math.Sqrt(float64(in)), nil
	})

	values, errs, _ := squared.Collect()
	for _, v := range values {
		fmt.Println(v)
	}

	for _, e := range errs {
		perr := e.(*pipefn.PipeError)
		fmt.Printf("error: item=%v, reason=%s\n", perr.Item, perr.Reason)
	}

}

Output:
1
4
error: item=-2, reason=NaN

func Zip

func Zip[Left, Right any](left Pipe[Left], right Pipe[Right]) Pipe[Tuple[Left, Right]]

Zip combines two pipes into a single Pipe that emits pairs of values, where each pair consists of one value from the left pipe and one value from the right pipe.

Values are produced in lockstep: Zip waits until both pipes have emitted a value, then emits a Tuple containing those values. The relative ordering within each input pipe is preserved.

The resulting pipe terminates as soon as either input pipe is exhausted. Any remaining values in the other pipe are discarded.

Zip may consume more values from its input pipes than are ultimately emitted. In particular, it may read ahead from one pipe even if the other pipe has already terminated.

Transformation errors from the input pipes are propagated unchanged.

After the stream has been fully consumed, Stream.Err() returns the first terminal error encountered from either input pipe, if any.

Example

package main

import (
	"fmt"

	"github.com/KasperOmsK/pipefn"
)

func main() {
	p1 := pipefn.FromSlice([]int{1, 2, 3, 4})
	p2 := pipefn.FromSlice([]string{"a", "b", "c"})

	zipped := pipefn.Zip(p1, p2)

	values, _, _ := zipped.Collect()

	for _, v := range values {
		fmt.Printf("left=%d, right=%s\n", v.LValue, v.RValue)
	}

}

Output:
left=1, right=a
left=2, right=b
left=3, right=c

func (*Pipe[T]) Collect

func (p *Pipe[T]) Collect() ([]T, []error, error)

Collect consumes the entire Pipe and returns all emitted values, all pipeline errors, and the terminal error of the value stream, if any.

Values are collected in the order they are produced by the Pipe. Errors are collected in the order they are emitted.

Collect fully drains the Pipe; after calling it, the Pipe cannot be consumed again.

func (*Pipe[T]) CollectValues

func (p *Pipe[T]) CollectValues() ([]T, error)

CollectValues consumes the entire Pipe and returns all emitted values and the terminal error of the value stream, if any.

Values are collected in the order they are produced by the Pipe. Pipeline errors are discarded.

CollectValues fully drains the Pipe; after calling it, the Pipe cannot be consumed again.

func (*Pipe[T]) ForEach

func (p *Pipe[T]) ForEach(consumeFn func(item T), errorFn func(err error)) error

ForEach consumes all values from the Pipe, invoking consumeFn for each.

Errors produced by the pipe are forwarded to errorFn.

consumeFn and errorFn are executed concurrently. If shared state is accessed, the caller is responsible for ensuring proper synchronization.

ForEach blocks until the pipe has been fully consumed and all pipeline errors have been delivered to errorFn. It then returns the terminal error reported by the value stream, if any.

Like for Results, calling ForEach is a destructive operation: any subsequent call to ForEach (or Results, ForEach, Collect etc.) will return an empty stream whose Err() method returns ErrAlreadyConsumed.

func (*Pipe[T]) Results

func (p *Pipe[T]) Results() (values Stream[T], errs <-chan error)

Results returns a Stream that yields the values produced by p, along with a channel that emits errors produced by the pipeline.

The errors channel contains only transformation errors produced by pipeline stages (for example via TryMap). Terminal error of the underlying Stream are not sent through this channel and must instead be checked via values.Err() after iteration completes.

Callers must drain the errors channel. If errors are not consumed, stages in the pipeline that attempt to emit errors may block.

Callers that wish to consume the pipeline without handling errors may use p.Values() instead.

All errors emitted by the errors channel will be of type *PipeError.

Results is a consumption method. See Pipe documentation for consumption semantics.

Typical usage:

values, errs := p.Results()

// drain errs
done := make(chan struct{})
go func() {
    defer close(done)
    for err := range errs {
        perr := err.(*PipeError)
        log.Printf("pipeline error: item %v: %s", perr.Item, perr.Reason)
    }
}()

// iterate over values
for v := range values.Seq() {
    process(v)
}

if err := values.Err() != nil {
    // handle terminal error (e.g. rollback or discard processed work)
}

<-done

func (*Pipe[T]) Tap

func (p *Pipe[T]) Tap(tapFn func(T))

Tap modifies p so that tapFn is called for each element that passes through p. The tap function is called before the element is yielded.

Calling Tap multiple times will chain tap functions in the order they were added.

Tap panics if tapFn is nil.

func (*Pipe[T]) Values

func (p *Pipe[T]) Values() Stream[T]

Values returns the sequence of values produced by p.

Values behaves like Results, except that the error sequence is consumed internally and discarded. This allows callers who do not care about errors to ignore them while still ensuring the pipe completes correctly.

Values is a consumption method. See Pipe documentation for consumption semantics.

Typical usage:

stream := p.Values()
for v := range stream.Seq() {
	process(v)
}

if err := stream.Err(); err != nil{
	// handle err
}

type PipeError

type PipeError struct {
	// The item that generated the error
	Item any
	// The actual error
	Reason error
}

PipeError represents an error that occured inside a pipeline

func (*PipeError) Error

func (pe *PipeError) Error() string

func (*PipeError) Unwrap

func (pe *PipeError) Unwrap() error

type Predicate

type Predicate[T any] func(item T) bool

Predicate represents a filtering function that returns true when the provided value should be included in the output stream.

type Stream

type Stream[T any] interface {
	// Seq returns the iterator that yields elements of the stream.
	Seq() iter.Seq[T]

	// Err returns the terminal error encountered while producing the stream.
	// It returns nil if the stream completed successfully. Err should be
	// checked only after the sequence returned by Seq has been fully consumed.
	Err() error
}

Stream represents a generic data stream.

A Stream produces a sequence of values of type T and may terminate with an error. Consumers iterate over the values using the Seq method and must check Err after iteration completes to determine whether the stream ended normally or due to an error.

Typical usage:

s := getStream()
for v := range s.Seq() {
    // process v
}
if err := s.Err(); err != nil {
    // handle stream error
}

The sequence returned by Seq should be consumed before calling Err to ensure that the stream has fully completed.

type TryMapFunc

type TryMapFunc[In, Out any] func(in In) (Out, error)

TryMapFunc is a mapping function that may return an error.

Errors are added to the Pipe's error channel while successful values continue through the pipeline.

type Tuple

type Tuple[Left, Right any] struct {
	LValue Left
	RValue Right
}