flow

README

flow

Implement the ends, flow owns the pipes. You write a Producer, a Processor and a Consumer (or just pass closures); flow provides the concurrency, ordering, back-pressure and fail-fast cancellation. Zero dependencies.

double := flow.Map(func(n int) int { return n * 2 })

var out []int
err := flow.Run(ctx, flow.Slice(1, 2, 3, 4), double, flow.Into(&out),
    flow.Workers(4), flow.Ordered())
// out == [2 4 6 8] — parallel, deterministic

No channels, no WaitGroup, no fan-in wiring in your code — just single-item logic and a couple of options.

The idea

A pipeline is always the same shape: a producer at one end, a consumer at the other, work in the middle.

flowchart LR
    P(["Producer"]) --> R
    subgraph R ["flow.Run — pool · ordering · back-pressure"]
      direction LR
      W1["Processor"]
      W2["Processor"]
      W3["Processor"]
    end
    R --> C(["Consumer"])

flow inverts control: instead of chaining channel operators yourself, you implement three small interfaces and hand them to Run, which owns the infrastructure. The design is inspired by Elixir's GenStage, adapted to idiomatic Go.

Install

go get github.com/pabloos/flow

Requires Go 1.23+.

The three interfaces

type Producer[T any]     interface { Produce(ctx context.Context, emit func(T) error) error }
type Processor[I, O any] interface { Process(ctx context.Context, in I, emit func(O) error) error }
type Consumer[T any]     interface { Consume(ctx context.Context, v T) error }

Implement them on a struct for stateful/complex ends, or pass a closure via the *Func adapters (the http.HandlerFunc trick) for quick ones:

// A struct end carries state; the consumer runs single-threaded, so no locks.
type Sum struct{ total int }

func (s *Sum) Consume(ctx context.Context, n int) error {
    s.total += n
    return nil
}

One Processor is Map, Filter and FlatMap — it depends on how many times you emit:

flow.ProcessorFunc[int, int](func(ctx context.Context, n int, emit func(int) error) error {
    if n < 0 { return nil }          // Filter: emit nothing
    if err := emit(n); err != nil {  // Map: emit one
        return err
    }
    return emit(n * 10)              // FlatMap: emit more
})

For the common cases, terse constructors hide the emit closure (with fallible Try* variants):

flow.Map(func(n int) int { return n * 2 })
flow.Filter(func(n int) bool { return n > 0 })
flow.FlatMap(func(s string) []string { return strings.Fields(s) })

flow.TryMap(func(s string) (int, error) { return strconv.Atoi(s) })

Composition

Then composes processors with no channel between them — plain function composition, so multi-stage pipelines run in-process inside each worker:

parse := flow.TryMap(strconv.Atoi)                    // string -> int
scale := flow.Map(func(n int) float64 { return float64(n) * 1.5 })
proc  := flow.Then(parse, scale)                      // Processor[string, float64]

A complete program

Parse strings to ints, keep the positives and double them — in parallel, with the output order preserved:

package main

import (
	"context"
	"fmt"
	"strconv"

	"github.com/pabloos/flow"
)

func main() {
	pipeline := flow.Then(
		flow.TryMap(strconv.Atoi),
		flow.Then(
			flow.Filter(func(n int) bool { return n > 0 }),
			flow.Map(func(n int) int { return n * 2 }),
		),
	)

	var out []int
	err := flow.Run(context.Background(),
		flow.Slice("1", "-2", "3", "4"),
		pipeline,
		flow.Into(&out),
		flow.Workers(4), flow.Ordered(),
	)
	if err != nil {
		panic(err)
	}
	fmt.Println(out) // [2 6 8]
}

More runnable examples live in the package docs (they run in CI, so they never go stale).

Concurrency and ordering

Both are options on Run — your processor code stays single-item and oblivious:

err := flow.Run(ctx, producer, proc, consumer,
    flow.Workers(8),    // pool size (parallelism)
    flow.Ordered(),     // reconstruct input order at the consumer
    flow.Prefetch(64),  // in-flight window: let the producer run ahead
)

Ordered() restores the original input order even though workers finish at different times (and it survives FlatMap sibling expansion). Without it, outputs are consumed as they arrive. The consumer is always called from a single goroutine, so it needs no locks.

Keyed ordering

RunPartitioned routes every input to a worker by key, so all values with the same key go to the same worker and are processed in order — a given key is never processed concurrently. Use it for per-key ordering or processors that keep per-key state:

err := flow.RunPartitioned(ctx, producer, proc, consumer,
    func(e Event) string { return e.UserID }, // same user -> same worker, in order
    flow.Workers(8),
)

Combine it with Ordered() to also restore the global input order at the consumer; on its own it only guarantees order within each key.

Batteries-included ends

flow.Slice(1, 2, 3)          // Producer from values
flow.FromSeq(seq)            // Producer from a Go 1.23 iter.Seq

flow.Into(&results)          // Consumer that collects into a slice
flow.Each(func(v T) error)   // Consumer that runs a func

Errors and cancellation

The first error from any end — producer, processor or consumer — cancels the whole pipeline (fail-fast) and is returned by Run:

err := flow.Run(ctx, prod, flow.TryMap(risky), sink)
if err != nil { /* the pipeline stopped at the first failure */ }

Back-pressure is implicit: emit blocks while the pipeline is saturated, all the way back to the producer.

Prior art

• destel/rill — composable channel operators with ordered variants; a great fit if you like the operator-chaining style.
• reugn/go-streams — Source/Flow/Sink DSL with many connectors (Kafka, NATS, …).
• Elixir GenStage — the demand-driven inspiration for the "implement the ends" model.

flow's niche is the interface-driven ergonomics: you plug in the ends, it owns the pipes.

History

flow began as the generics rewrite of GoPipelines, a pre-generics study of the Go pipelines pattern. Its lineage is preserved in the repo:

• v1-legacy tag — the original int-only pattern.
• channel-api tag — the generics rewrite with an explicit channel/Stage API.
• main — the current interface-driven design.

Limitations

• Back-pressure is by blocking emit, not demand-driven like GenStage.
• Multi-stage type changes compose two at a time (Then), a Go generics constraint — turned into in-process, channel-free composition.

License

MIT

Documentation

Overview

Package flow is an interface-driven library for concurrent pipelines: you implement the ends — Producer, Processor and Consumer (or pass a closure via the *Func adapters) — and flow owns the concurrency, ordering, back-pressure and fail-fast cancellation.

Inspired by Elixir's GenStage, but with push+blocking back-pressure and streaming reconstruction of the global input order.

Example (StructConsumer)

Ends can be structs when they carry state; the consumer is called from a single goroutine, so no locking is needed.

package main

import (
	"context"
	"fmt"

	"github.com/pabloos/flow"
)

// sum is a stateful Consumer implemented on a struct rather than a closure.
type sum struct{ total int }

func (s *sum) Consume(ctx context.Context, n int) error {
	s.total += n
	return nil
}

func main() {
	total := &sum{}
	err := flow.Run(context.Background(),
		flow.Slice(1, 2, 3, 4),
		flow.Map(func(n int) int { return n }),
		total,
		flow.Workers(4),
	)
	if err != nil {
		fmt.Println("error:", err)
	}
	fmt.Println(total.total)
}

Output:
10

Index

• func Run[I, O any](ctx context.Context, p Producer[I], proc Processor[I, O], c Consumer[O], ...) error
• func RunPartitioned[I, O any, K comparable](ctx context.Context, p Producer[I], proc Processor[I, O], c Consumer[O], ...) error
• type Consumer
  • func Each[T any](fn func(T) error) Consumer[T]
  • func Into[T any](dst *[]T) Consumer[T]
• type ConsumerFunc
  • func (f ConsumerFunc[T]) Consume(ctx context.Context, v T) error
• type Option
  • func Ordered() Option
  • func Prefetch(n int) Option
  • func Workers(n int) Option
• type Processor
  • func Filter[T any](pred func(T) bool) Processor[T, T]
  • func FlatMap[I, O any](fn func(I) []O) Processor[I, O]
  • func Map[I, O any](fn func(I) O) Processor[I, O]
  • func Then[A, B, C any](p1 Processor[A, B], p2 Processor[B, C]) Processor[A, C]
  • func TryFilter[T any](pred func(T) (bool, error)) Processor[T, T]
  • func TryFlatMap[I, O any](fn func(I) ([]O, error)) Processor[I, O]
  • func TryMap[I, O any](fn func(I) (O, error)) Processor[I, O]
• type ProcessorFunc
  • func (f ProcessorFunc[I, O]) Process(ctx context.Context, in I, emit func(O) error) error
• type Producer
  • func FromSeq[T any](seq iter.Seq[T]) Producer[T]
  • func Slice[T any](xs ...T) Producer[T]
• type ProducerFunc
  • func (f ProducerFunc[T]) Produce(ctx context.Context, emit func(T) error) error

Examples

• Package (StructConsumer)
• ProcessorFunc
• Run
• RunPartitioned
• Then

Functions

func Run

func Run[I, O any](ctx context.Context, p Producer[I], proc Processor[I, O], c Consumer[O], opts ...Option) error

Run wires producer -> pool of processors -> consumer and owns the concurrency, ordering, back-pressure and fail-fast cancellation. It returns the first error from any end, or nil.

Example

The basic shape: a producer, a processor, a consumer, run in parallel with the output order preserved.

package main

import (
	"context"
	"fmt"

	"github.com/pabloos/flow"
)

func main() {
	err := flow.Run(context.Background(),
		flow.Slice(1, 2, 3, 4),
		flow.Map(func(n int) int { return n * 2 }),
		flow.Each(func(n int) error {
			fmt.Println(n)
			return nil
		}),
		flow.Workers(4),
		flow.Ordered(),
	)
	if err != nil {
		fmt.Println("error:", err)
	}
}

Output:
2
4
6
8

func RunPartitioned

func RunPartitioned[I, O any, K comparable](ctx context.Context, p Producer[I], proc Processor[I, O], c Consumer[O], key func(I) K, opts ...Option) error

RunPartitioned is like Run but routes every input to a worker by key: all values with the same key go to the same worker and are processed in order, so a given key is never processed concurrently. Use it for per-key ordering or for processors that keep per-key state.

Combine with Ordered to also restore the global input order at the consumer; on its own, RunPartitioned only guarantees order within each key.

Example

RunPartitioned keeps each key's values on one worker, in order. With Ordered the global input order is restored too.

package main

import (
	"context"
	"fmt"

	"github.com/pabloos/flow"
)

func main() {
	type event struct {
		user string
		n    int
	}
	events := []event{{"a", 1}, {"b", 1}, {"a", 2}, {"b", 2}}

	err := flow.RunPartitioned(context.Background(),
		flow.Slice(events...),
		flow.Map(func(e event) string { return fmt.Sprintf("%s#%d", e.user, e.n) }),
		flow.Each(func(s string) error {
			fmt.Println(s)
			return nil
		}),
		func(e event) string { return e.user },
		flow.Workers(4),
		flow.Ordered(),
	)
	if err != nil {
		fmt.Println("error:", err)
	}
}

Output:
a#1
b#1
a#2
b#2

Types

type Consumer

type Consumer[T any] interface {
	Consume(ctx context.Context, v T) error
}

Consumer receives each output value. It is never called concurrently, so implementations need no synchronization.

func Each

func Each[T any](fn func(T) error) Consumer[T]

Each runs fn for every output.

func Into

func Into[T any](dst *[]T) Consumer[T]

Into collects outputs into the given slice. Safe without locks: the consumer is called from a single goroutine.

type ConsumerFunc

type ConsumerFunc[T any] func(ctx context.Context, v T) error

func (ConsumerFunc[T]) Consume

func (f ConsumerFunc[T]) Consume(ctx context.Context, v T) error

type Option

type Option func(*config)

Option configures a Run.

func Ordered

func Ordered() Option

Ordered reconstructs the original input order at the consumer, even with a worker pool. Without it, outputs are consumed in arrival order.

func Prefetch

func Prefetch(n int) Option

Prefetch sets how many items may be queued ahead of the workers (the in-flight window), letting the producer run ahead instead of lock-stepping with the pool. Default 0 (unbuffered: strict lock-step).

func Workers

func Workers(n int) Option

Workers sets the size of the processor pool (parallelism). Default 1.

type Processor

type Processor[I, O any] interface {
	Process(ctx context.Context, in I, emit func(O) error) error
}

Processor turns one input value into zero or more outputs via emit. Emitting once is Map, emitting conditionally is Filter, emitting many is FlatMap.

func Filter

func Filter[T any](pred func(T) bool) Processor[T, T]

Filter builds a Processor that keeps values for which pred is true.

func FlatMap

func FlatMap[I, O any](fn func(I) []O) Processor[I, O]

FlatMap builds a Processor that expands each value into zero or more outputs.

func Map

func Map[I, O any](fn func(I) O) Processor[I, O]

Map builds a Processor from a plain transform. Use TryMap if it can fail.

func Then

func Then[A, B, C any](p1 Processor[A, B], p2 Processor[B, C]) Processor[A, C]

Then composes two processors with no channel between them: it is plain function composition through emit, so multi-stage pipelines run in-process inside each worker.

Example

A realistic multi-stage pipeline composed with Then: parse, filter, double.

package main

import (
	"context"
	"fmt"
	"strconv"

	"github.com/pabloos/flow"
)

func main() {
	pipeline := flow.Then(
		flow.TryMap(strconv.Atoi), // string -> int
		flow.Then(
			flow.Filter(func(n int) bool { return n > 0 }), // keep positives
			flow.Map(func(n int) int { return n * 2 }),     // double
		),
	)

	var out []int
	err := flow.Run(context.Background(),
		flow.Slice("1", "-2", "3", "4"),
		pipeline,
		flow.Into(&out),
		flow.Workers(4),
		flow.Ordered(),
	)
	if err != nil {
		fmt.Println("error:", err)
	}
	fmt.Println(out)
}

Output:
[2 6 8]

func TryFilter

func TryFilter[T any](pred func(T) (bool, error)) Processor[T, T]

TryFilter is Filter with a fallible predicate.

func TryFlatMap

func TryFlatMap[I, O any](fn func(I) ([]O, error)) Processor[I, O]

TryFlatMap is FlatMap with a fallible expansion.

func TryMap

func TryMap[I, O any](fn func(I) (O, error)) Processor[I, O]

TryMap builds a Processor from a fallible transform; a returned error aborts the pipeline (fail-fast).

type ProcessorFunc

type ProcessorFunc[I, O any] func(ctx context.Context, in I, emit func(O) error) error

Example

One Processor can expand each input into several outputs (FlatMap) by emitting more than once.

package main

import (
	"context"
	"fmt"
	"strings"

	"github.com/pabloos/flow"
)

func main() {
	split := flow.ProcessorFunc[string, string](func(ctx context.Context, line string, emit func(string) error) error {
		for _, w := range strings.Fields(line) {
			if err := emit(w); err != nil {
				return err
			}
		}
		return nil
	})

	var words []string
	err := flow.Run(context.Background(),
		flow.Slice("hello world", "goodbye"),
		split,
		flow.Into(&words),
		flow.Ordered(),
	)
	if err != nil {
		fmt.Println("error:", err)
	}
	fmt.Println(words)
}

Output:
[hello world goodbye]

func (ProcessorFunc[I, O]) Process

func (f ProcessorFunc[I, O]) Process(ctx context.Context, in I, emit func(O) error) error

type Producer

type Producer[T any] interface {
	Produce(ctx context.Context, emit func(T) error) error
}

Producer emits values into the pipeline via emit until it is done or errors.

func FromSeq

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

FromSeq produces the values of a Go 1.23 iterator.

func Slice

func Slice[T any](xs ...T) Producer[T]

Slice produces the given values in order.

type ProducerFunc

type ProducerFunc[T any] func(ctx context.Context, emit func(T) error) error

func (ProducerFunc[T]) Produce

func (f ProducerFunc[T]) Produce(ctx context.Context, emit func(T) error) error