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.

Profiling & bottlenecks

Because flow owns the infrastructure, it can profile it for you — with no instrumentation in your code. Pass Observe to fill a Report that pinpoints the bottleneck:

var rep flow.Report
err := flow.Run(ctx, prod, proc, cons, flow.Workers(8), flow.Observe(&rep))
fmt.Println(&rep)

flow report — wall 104ms · 8 workers
  producer   3000 items   blocked 101ms
  processor  3000 in / 3000 out   busy 810ms  idle 1ms  blocked 4ms  · 98% util
             latency  p50 120µs  p95 3.004ms  p99 3.025ms  max 3.14ms
             balance  busy 101ms–103ms across 8 workers
  consumer   3000 items   busy 0s
  → bottleneck: PROCESSOR
    processor-bound: raise Workers() or optimize Process

flow triangulates by where the workers spend their time — idle (producer too slow), busy (the work itself) or blocked (consumer too slow) — so rep.Bottleneck is one of StageProducer / StageProcessor / StageConsumer, with matching Advice. The latency line exposes the Process tail (here p50 is 120µs but p95 jumps to 3ms — a spike a mean would hide), and balance flags uneven workers; the full data is in rep.ProcessLatency and rep.PerWorker. Observe is opt-in; with it off, the hot path pays nothing.

Streaming metrics

Observe is a one-shot snapshot; for live metrics during a run — and for long-lived pipelines that never "end" — implement Meter and pass WithMeter. flow records the counters produced / processed / emitted / consumed and the process_latency histogram to your instruments:

err := flow.Run(ctx, prod, proc, cons, flow.Workers(8), flow.WithMeter(m))

Meter is deliberately tiny (instrument-based, à la OpenTelemetry) so the core stays zero-dependency; concrete exporters (Prometheus, OpenTelemetry) live in separate submodules under flow/x/….

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.

Roadmap

Observability grows without breaking the core's zero-dependency promise:

• In core (stdlib only): Report percentiles and per-worker breakdown (done, v0.4.0) and the Meter streaming hook (done, v0.5.0).
• Separate modules (own go.mod, opt-in) — next: concrete exporters implementing Meter under flow/x/… (Prometheus, then OpenTelemetry), so importing the core never pulls their dependencies.

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 Counter
• type Histogram
• type Meter
• type Option
  • func Observe(dst *Report) Option
  • func Ordered() Option
  • func Prefetch(n int) Option
  • func WithMeter(m Meter) 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
• type Report
  • func (r *Report) String() string
  • func (r *Report) Util() float64
• type Stage
  • func (s Stage) String() string
• type StageLatency
• type WorkerStat

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 Counter

type Counter interface {
	Add(n int64)
}

Counter is a monotonically increasing count.

type Histogram

type Histogram interface {
	Record(d time.Duration)
}

Histogram records a distribution of durations.

type Meter

type Meter interface {
	Counter(name string) Counter
	Histogram(name string) Histogram
}

Meter is flow's streaming-metrics extension point. Implement it in a separate module — so the core stays zero-dependency — to forward live measurements to a backend such as Prometheus or OpenTelemetry, and enable it with WithMeter.

The engine asks the Meter for its instruments once when a run starts, then updates them on the hot path. Instrument methods are called from many worker goroutines at once, so implementations must be safe for concurrent use.

Experimental: the observability API may change in a future v0.x release.

type Option

type Option func(*config)

Option configures a Run.

func Observe

func Observe(dst *Report) Option

Observe fills dst with a performance Report after the run: per-stage timings, counts and a diagnosed bottleneck. Opt-in — it adds timing to the hot path, so leave it off in production.

Experimental: the observability API (Observe and Report) may change in a future v0.x release.

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 WithMeter

func WithMeter(m Meter) Option

WithMeter streams live measurements to m during the run: the counters "produced", "processed", "emitted" and "consumed", and the histogram "process_latency". Opt-in; with no Meter the hot path pays nothing.

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

type Report

type Report struct {
	Wall    time.Duration
	Workers int

	Produced  int64 // values emitted by the producer
	Processed int64 // inputs handled by the pool
	Emitted   int64 // outputs produced by the pool (fan-out with FlatMap)
	Consumed  int64 // values delivered to the consumer

	ProducerBlocked time.Duration // producer time blocked emitting (downstream full)
	WorkerBusy      time.Duration // time inside Process (summed)
	WorkerIdle      time.Duration // time waiting for input (producer slow)
	WorkerBlocked   time.Duration // time blocked delivering (consumer slow)
	ConsumerBusy    time.Duration // time inside Consume

	ProcessLatency StageLatency // per-item Process latency distribution
	PerWorker      []WorkerStat // per-worker breakdown

	Bottleneck Stage
	Advice     string
}

Report is a snapshot of a run's performance, filled in when Observe is set. Aggregate processor timings are summed across all workers; PerWorker holds the per-worker breakdown.

Experimental: this type may change in a future v0.x release.

func (*Report) String

func (r *Report) String() string

func (*Report) Util

func (r *Report) Util() float64

Util is the fraction of available worker-time spent doing work (0..1).

type Stage

type Stage int

Stage identifies a pipeline stage.

const (
	StageProducer Stage = iota
	StageProcessor
	StageConsumer
)

func (Stage) String

func (s Stage) String() string

type StageLatency

type StageLatency struct {
	P50, P95, P99, Max time.Duration
}

StageLatency holds the per-item latency distribution of Process. Percentiles are computed from a bounded reservoir sample, so they are approximate; Max is exact.

type WorkerStat

type WorkerStat struct {
	Processed           int64
	Busy, Idle, Blocked time.Duration
}

WorkerStat is one worker's share of the work, for spotting imbalance.