Operator Reference

This document covers every operator in go-kitsune. Each entry shows the exact Go signature, what the operator does, when to use or avoid it, which StageOption values apply, and a minimal working example.

Free functions (Map, FlatMap, Batch, …) can change the element type as items flow through. Methods (.Filter, .Take, .Skip, …) preserve the type. This split is a Go generic constraint: methods cannot introduce their own type parameters, so anything that changes Pipeline[A] to Pipeline[B] must be a free function.

Quick reference

Jump directly to any operator. See Contents for a grouped view.

Contents

1. Sources
2. 1:1 Transforms
3. 1:N Expansion
4. Filtering & Selection
5. Stateful Transforms
6. Batching & Windowing
7. Fan-Out & Fan-In
8. Enrichment
9. Aggregation & Collection
10. Time-Based Operators
11. Resilience
12. Utility & Metadata
13. Terminal Operators
14. Stage Composition
15. Error Handling Options
16. Stage Options Reference

Sources

Sources are the entry point of every pipeline. They have no input pipeline; they produce items from an external data source, a collection, or a generator function.

FromSlice

func FromSlice[T any](items []T) *Pipeline[T]

Creates a pipeline that emits each element of items in order, then closes. The slice is captured by reference at construction time; do not modify it after calling FromSlice.

When to use: Tests, small fixed datasets, or any time you already have all the data in memory.

When not to use: Large datasets where you want to stream lazily; use Generate or FromIter instead.

Options: none (sources take no StageOption).

nums := kitsune.FromSlice([]int{1, 2, 3, 4, 5})
result, _ := nums.Collect(ctx)
// result == [1, 2, 3, 4, 5]

From

func From[T any](src <-chan T) *Pipeline[T]

Wraps an existing channel as a pipeline source. The pipeline completes when src is closed.

When to use: Bridging existing channel-based code (Kafka consumer channels, os/signal channels, etc.) into a kitsune pipeline.

When not to use: When you control the producer; NewChannel or Generate give you better lifecycle management.

Options: none.

ch := make(chan Event, 64)
go kafkaConsumer(ch) // fills ch and closes it when done
p := kitsune.From(ch)

Generate

func Generate[T any](fn func(ctx context.Context, yield func(T) bool) error) *Pipeline[T]

Creates a push-based source from a generator function. Call yield(item) for each item to emit; yield returns false when the pipeline is shutting down. Return nil on clean completion or an error to propagate it.

Generate is the lowest-level source primitive; all other sources are implemented with it. The context passed to fn is cancelled when the pipeline shuts down, so any blocking I/O inside fn (long-poll RPCs, database cursors) is interrupted cleanly.

When to use: Paginated APIs, database cursors, WebSocket feeds, or any external source you can drive with a loop.

Options: none.

pages := kitsune.Generate(func(ctx context.Context, yield func(Page) bool) error {
    cursor := ""
    for {
        page, next, err := api.Fetch(ctx, cursor)
        if err != nil {
            return err
        }
        if !yield(page) {
            return nil
        }
        if next == "" {
            return nil
        }
        cursor = next
    }
})

FromIter

func FromIter[T any](seq iter.Seq[T]) *Pipeline[T]

Wraps a Go 1.23 iter.Seq[T] iterator as a pipeline source. Cancellation is checked between items.

When to use: When you have an iter.Seq[T] from slices.Values, maps.Keys, or a custom iterator and want to plug it into a pipeline.

Options: none.

import "slices"

p := kitsune.FromIter(slices.Values([]string{"a", "b", "c"}))

NewChannel / Channel[T]

func NewChannel[T any](buffer int) *Channel[T]
func (c *Channel[T]) Source() *Pipeline[T]
func (c *Channel[T]) Send(ctx context.Context, item T) error
func (c *Channel[T]) TrySend(item T) bool
func (c *Channel[T]) Close()

A thread-safe, push-based source for external producers. Create with NewChannel, obtain the pipeline with Source(), push items with Send or TrySend, and call Close when no more items will be sent. Close is idempotent. Send blocks if the buffer is full; TrySend returns false immediately instead of blocking.

When to use: HTTP handlers, gRPC streams, or any external goroutine that needs to feed items into a running pipeline without knowing about kitsune internals.

Options: none.

ch := kitsune.NewChannel[Order](64)
p := ch.Source()

// In an HTTP handler (separate goroutine):
go func() {
    http.HandleFunc("/order", func(w http.ResponseWriter, r *http.Request) {
        var o Order
        json.NewDecoder(r.Body).Decode(&o)
        ch.Send(r.Context(), o)
    })
}()

// When the server shuts down:
ch.Close()

Generate vs Channel[T]

Both Generate and Channel[T] bridge external code into a pipeline, but they suit different producer shapes.

Choose Generate when the producer is a loop you can express inline: paginated REST APIs, database cursors, polling a queue, walking a filesystem. The pipeline owns the loop and shuts it down cleanly via context cancellation.

Choose Channel[T] when items arrive from goroutines the pipeline does not own: HTTP handlers, gRPC stream handlers, library callbacks, fan-in from multiple producer goroutines. Producers stay decoupled from kitsune internals and only need a Send call.

If you find yourself starting a goroutine inside Generate just to call yield, use Channel[T] instead. If you find yourself wrapping Channel[T] in a single-goroutine loop, use Generate instead.

Ticker

func Ticker(d time.Duration, opts ...StageOption) *Pipeline[time.Time]

Emits the current time.Time at regular intervals. The first tick fires after d. The pipeline runs indefinitely until the context is cancelled or a downstream operator (like Take) stops it.

When to use: Periodic polling, heartbeats, scheduled work triggered by time.

Options: WithClock, WithName.

// Poll every 30 seconds; stop after 10 polls.
ticks := kitsune.Ticker(30 * time.Second).Take(10)

Timer

func Timer[T any](delay time.Duration, fn func() T, opts ...StageOption) *Pipeline[T]

Emits exactly one value after delay by calling fn, then closes. If the context is cancelled before delay elapses, no value is emitted.

When to use: Deferred notifications, one-shot scheduled events, timeouts that produce a sentinel value.

Options: WithClock, WithName.

// Emit a "ping" string after 5 seconds.
ping := kitsune.Timer(5*time.Second, func() string { return "ping" })

Unfold

func Unfold[S, T any](seed S, fn func(S) (T, S, bool)) *Pipeline[T]

Generates a stream by repeatedly applying fn to an accumulator. fn receives the current state and returns (value, nextState, stop). When stop is true, the stream ends without emitting the value.

When to use: Sequences derived from recurrences (Fibonacci, geometric series, tree traversals) where each step depends on the last.

Options: none.

// Fibonacci sequence
fib := kitsune.Unfold([2]int{0, 1}, func(s [2]int) (int, [2]int, bool) {
    return s[0], [2]int{s[1], s[0] + s[1]}, false
}).Take(8)
// emits: 0, 1, 1, 2, 3, 5, 8, 13

Iterate

func Iterate[T any](seed T, fn func(T) T) *Pipeline[T]

Creates an infinite stream starting with seed where each subsequent value is produced by applying fn to the previous one. Use Take or TakeWhile to bound it.

Options: none.

kitsune.Iterate(1, func(n int) int { return n * 2 }).Take(5)
// emits: 1, 2, 4, 8, 16

Repeatedly

func Repeatedly[T any](fn func() T) *Pipeline[T]

Creates an infinite stream by calling fn on each iteration. Use Take or TakeWhile to bound it.

When to use: Sampling a random number generator, reading from a sensor, generating UUIDs.

Options: none.

kitsune.Repeatedly(rand.Int64).Take(100)

Cycle

func Cycle[T any](items []T) *Pipeline[T]

Creates an infinite stream that repeatedly loops over items. Panics if items is empty.

Options: none.

kitsune.Cycle([]string{"a", "b", "c"}).Take(7)
// emits: "a","b","c","a","b","c","a"

Empty

func Empty[T any]() *Pipeline[T]

Returns a Pipeline that completes immediately with no items. Useful as an identity element in pipeline composition and as a base case in tests.

Merge(Empty[T](), p) behaves identically to p for any pipeline p. Amb(Empty[T](), p) is not the same: Amb forwards whichever emits first, and Empty completes before emitting, so the winner is p.

Options: none.

// Use as a no-op source in conditional pipelines:
var src *kitsune.Pipeline[Event]
if cond {
    src = realSource()
} else {
    src = kitsune.Empty[Event]()
}

See also: Never (blocks indefinitely), FromSlice with a nil slice.

Never

func Never[T any]() *Pipeline[T]

Returns a Pipeline that never emits any items and never completes until the context is cancelled. The absorbing element for Amb: Amb(Never[T](), p) always forwards from p.

Useful as a placeholder in tests that assert on other branches, or to keep a merged pipeline alive while other branches are active.

Options: none.

// In a test: ensure the other branch wins the race.
winner := kitsune.Amb(
    func() *kitsune.Pipeline[int] { return kitsune.Never[int]() },
    func() *kitsune.Pipeline[int] { return kitsune.FromSlice([]int{1, 2, 3}) },
)

See also: Empty (completes immediately), Amb.

Concat

func Concat[T any](factories ...func() *Pipeline[T]) *Pipeline[T]

Runs each pipeline factory in order: all items from factory[0] are emitted before factory[1] starts. Factories are called lazily; each one is invoked only after the previous pipeline has fully completed. Accepts factories (not *Pipeline values directly) so each run creates a fresh pipeline graph.

When to use: Sequential sources that must not overlap, such as first emitting a header, then body rows, then a footer.

Options: none.

kitsune.Concat(
    func() *kitsune.Pipeline[int] { return kitsune.FromSlice([]int{1, 2}) },
    func() *kitsune.Pipeline[int] { return kitsune.FromSlice([]int{3, 4}) },
)
// emits: 1, 2, 3, 4

Amb

func Amb[T any](factories ...func() *Pipeline[T]) *Pipeline[T]

Subscribes to all factories concurrently and forwards items exclusively from whichever factory emits first, immediately cancelling all others. If no factory emits before the context is cancelled, the pipeline produces no items.

When to use: Hedged requests: try a primary and a replica simultaneously, use whichever responds first.

Options: none.

result := kitsune.Amb(
    func() *kitsune.Pipeline[Result] { return fetchFromPrimary(ctx) },
    func() *kitsune.Pipeline[Result] { return fetchFromReplica(ctx) },
)

Catch

func Catch[T any](p *Pipeline[T], fn func(error) *Pipeline[T]) *Pipeline[T]

Runs p normally. If p returns a non-nil, non-context error, fn is called with the error and the returned fallback pipeline is subscribed; items already emitted by p are kept and the fallback's items follow. If p completes without error, the fallback is never started.

When to use: Streaming fallback to a secondary source when the primary fails.

Options: none.

kitsune.Catch(primaryFeed, func(err error) *kitsune.Pipeline[Event] {
    log.Printf("primary failed (%v); switching to backup", err)
    return backupFeed()
})

Retry

func Retry[T any](p *Pipeline[T], pol RetryPolicy) *Pipeline[T]

Re-runs the entire pipeline p from scratch whenever it errors, according to pol. This is the right primitive for sources that must reconnect on failure: websocket tails, change-data-capture streams, long-poll HTTP. The correct response to a disconnect is to re-establish the connection and resume.

Items produced during any attempt (including partial output from a failed attempt) are forwarded downstream immediately; Retry does not buffer or replay. Downstream observes the concatenation of each attempt's output.

When to use: Persistent reconnect-on-drop semantics for a source pipeline. For per-item retries of a transformation function, use OnError(RetryMax(...)) instead.

Options: none. Configure behaviour via the RetryPolicy argument.

RetryPolicy fields:

Convenience constructors:

kitsune.RetryUpTo(n, backoff)   // at most n total attempts
kitsune.RetryForever(backoff)   // retry until context cancellation

// Reconnecting websocket tail: retry forever with exponential backoff.
kitsune.Retry(
    kitsune.Generate(websocketTail),
    kitsune.RetryForever(kitsune.ExponentialBackoff(100*time.Millisecond, 30*time.Second)),
)

// Retry up to 5 times, only on transient network errors.
kitsune.Retry(
    primaryFeed,
    kitsune.RetryUpTo(5, kitsune.FixedBackoff(time.Second)).
        WithRetryable(func(err error) bool { return errors.Is(err, io.ErrUnexpectedEOF) }).
        WithOnRetry(func(attempt int, err error, _ time.Duration) {
            log.Printf("retry %d after: %v", attempt+1, err)
        }),
)

Using

func Using[T, R any](
    acquire func(context.Context) (R, error),
    build func(R) *Pipeline[T],
    release func(R),
) *Pipeline[T]

Acquires a resource, builds a pipeline from it, and guarantees release is called exactly once when the pipeline exits, regardless of success, error, or cancellation. If acquire returns an error, no items are emitted and release is not called.

When to use: Database connections, file handles, or any resource that must be explicitly released when the pipeline finishes.

Options: none.

p := kitsune.Using(
    func(ctx context.Context) (*sql.Rows, error) {
        return db.QueryContext(ctx, "SELECT id, name FROM users")
    },
    func(rows *sql.Rows) *kitsune.Pipeline[User] {
        return kitsune.Generate(func(ctx context.Context, yield func(User) bool) error {
            for rows.Next() {
                var u User
                if err := rows.Scan(&u.ID, &u.Name); err != nil {
                    return err
                }
                if !yield(u) {
                    return nil
                }
            }
            return rows.Err()
        })
    },
    func(rows *sql.Rows) { rows.Close() },
)

1:1 Transforms

Each item in produces exactly one item out, potentially of a different type.

Map

func Map[I, O any](p *Pipeline[I], fn func(context.Context, I) (O, error), opts ...StageOption) *Pipeline[O]

Applies fn to each item, potentially changing the element type from I to O. The context passed to fn is cancelled if the pipeline shuts down; always use it for any I/O calls inside fn.

With Concurrency(n) > 1, fn is called from n goroutines in parallel. Results arrive in completion order unless you add Ordered(). The engine has a serial fast path that activates when concurrency is 1, there is no error handler override, and no hook; this path avoids all overhead per item.

When to use: Any 1:1 transformation: JSON parsing, struct conversion, external API calls.

When not to use: When you need to emit zero or multiple outputs per item; use Filter or FlatMap.

Options: Concurrency, Ordered, OnError, Buffer, Overflow, WithName, Timeout, Supervise, CacheBy.

// Parse log lines concurrently, preserving order.
parsed := kitsune.Map(lines, func(ctx context.Context, line string) (LogEntry, error) {
    return parseLine(line)
}, kitsune.Concurrency(8), kitsune.Ordered())

MapRecover

func MapRecover[I, O any](
    p *Pipeline[I],
    fn func(context.Context, I) (O, error),
    recover func(context.Context, I, error) O,
    opts ...StageOption,
) *Pipeline[O]

Applies fn to each item. If fn returns an error or panics, recover is called with the original item and the error to produce a fallback output value. The output pipeline always emits exactly one item per input; no items are dropped and the pipeline never halts on a per-item failure.

When to use: When you want a guaranteed-1:1 transform that substitutes a sentinel or default on failure, rather than propagating the error or dropping the item.

Options: Buffer, Timeout, WithName.

enriched := kitsune.MapRecover(events,
    func(ctx context.Context, e Event) (EnrichedEvent, error) {
        return enrich(ctx, e)
    },
    func(ctx context.Context, e Event, err error) EnrichedEvent {
        log.Printf("enrich failed for %v: %v", e.ID, err)
        return EnrichedEvent{Event: e, Enriched: false}
    },
)

MapResult

func MapResult[I, O any](
    p *Pipeline[I],
    fn func(context.Context, I) (O, error),
    opts ...StageOption,
) (*Pipeline[O], *Pipeline[ErrItem[I]])

Applies fn to each item and routes by outcome: successful outputs go to the first (ok) pipeline; failures go to the second (failed) pipeline as ErrItem[I] values containing the original input and the error. The pipeline never halts. Both output pipelines must be consumed.

ErrItem[I] is defined as:

type ErrItem[I any] struct {
    Item I
    Err  error
}

Unlike OnError(Skip()), failed items are not silently discarded; they are available for inspection, logging, or dead-letter storage. MapResult does not retry; wrap fn with your own retry loop if you need transient-failure handling before routing to the dead-letter branch.

When to use: When you want to process failures separately without halting the pipeline: audit trails, error reporting pipelines, reprocessing queues.

Options: Buffer, Timeout, WithName.

ok, failed := kitsune.MapResult(records, func(ctx context.Context, r Record) (Stored, error) {
    return db.Store(ctx, r)
})

runner, _ := kitsune.MergeRunners(
    ok.ForEach(func(_ context.Context, s Stored) error {
        metrics.Inc("stored")
        return nil
    }),
    failed.ForEach(func(_ context.Context, e kitsune.ErrItem[Record]) error {
        log.Printf("failed to store %v: %v", e.Item.ID, e.Err)
        return nil
    }),
)
runner.Run(ctx)

Timestamp

func Timestamp[T any](p *Pipeline[T], opts ...StageOption) *Pipeline[Timestamped[T]]

Tags each item with the wall-clock time it was observed at this stage. Emits Timestamped[T]{Value T; Time time.Time}.

Options: WithClock, WithName, Buffer.

ts := kitsune.Timestamp(events)
// emits: Timestamped[Event]{Value: e, Time: <now>}

TimeInterval

func TimeInterval[T any](p *Pipeline[T], opts ...StageOption) *Pipeline[TimedInterval[T]]

Tags each item with the elapsed duration since the previous item. The first item always has Elapsed == 0. Emits TimedInterval[T]{Value T; Elapsed time.Duration}. Always runs at Concurrency(1).

Options: WithClock, WithName, Buffer.

intervals := kitsune.TimeInterval(events)
// use intervals to detect stalls: if Elapsed > threshold, alert

StartWith

func StartWith[T any](p *Pipeline[T], items ...T) *Pipeline[T]

Prepends one or more items before the first item from p. The prefix is always emitted in full before p begins.

Options: none.

withHeader := kitsune.StartWith(rows, headerRow)

DefaultIfEmpty

func DefaultIfEmpty[T any](p *Pipeline[T], defaultVal T, opts ...StageOption) *Pipeline[T]

Forwards all items from p unchanged. If p completes without emitting any items, emits defaultVal once.

Options: Buffer, WithName.

result := kitsune.DefaultIfEmpty(searchResults, noResultsSentinel)

Intersperse

func Intersperse[T any](p *Pipeline[T], sep T, opts ...StageOption) *Pipeline[T]

Inserts sep between consecutive items. The separator is never emitted at the start or end of the stream.

Options: Buffer, WithName.

// 1,2,3 → 1,0,2,0,3
kitsune.Intersperse(nums, 0)

LiftPure / LiftFallible

func LiftPure[I, O any](fn func(I) O) func(context.Context, I) (O, error)
func LiftFallible[I, O any](fn func(I) (O, error)) func(context.Context, I) (O, error)

Adapter helpers that wrap a context-free or context-free-fallible function into the signature expected by Map, Filter, etc.

doubled := kitsune.Map(p, kitsune.LiftPure(func(n int) int { return n * 2 }))

Materialize / Dematerialize

func Materialize[T any](p *Pipeline[T]) *Pipeline[Notification[T]]
func Dematerialize[T any](p *Pipeline[Notification[T]], opts ...StageOption) *Pipeline[T]

Materialize converts each item and the terminal outcome of p into a Notification[T] value emitted on a single output pipeline. Dematerialize is the inverse: it unwraps a Notification[T] stream back into a plain T stream.

Notification[T] is a sum type:

type Notification[T any] struct {
    Value T
    Err   error
    Done  bool
}

Helper constructors and predicates are provided:

Materialize emission rules:

The key property: Materialize never propagates pipeline errors. They are encoded as the final notification instead, so downstream operators continue running. Context cancellation is not materialized; it exits the run immediately.

Dematerialize processing rules:

When to use: When you need to pass error events through operators that only handle T; for example, routing, logging, or filtering errors without halting the pipeline. The standard pattern:

// Classify notifications by outcome without halting on errors.
classified := kitsune.Map(
    kitsune.Materialize(src),
    func(_ context.Context, n kitsune.Notification[Event]) (TaggedEvent, error) {
        if n.IsError() {
            return TaggedEvent{Err: n.Err, Source: "pipeline"}, nil
        }
        if n.IsComplete() {
            return TaggedEvent{Done: true}, nil
        }
        return TaggedEvent{Event: n.Value}, nil
    },
)

Options (Dematerialize only): Buffer, Overflow, WithName.

Options (Materialize): none; it wraps the upstream run internally and cannot accept per-stage options.

See also: MapResult (routes errors to a separate pipeline without materialization), TapError (side-effect on terminal error), Catch (fallback pipeline on error).

1:N Expansion

These operators allow each input item to produce zero or more output items.

FlatMap

func FlatMap[I, O any](
    p *Pipeline[I],
    fn func(context.Context, I, func(O) error) error,
    opts ...StageOption,
) *Pipeline[O]

Transforms each input into zero or more outputs. fn calls yield(item) for each output it wants to emit; yield returns an error if the pipeline is shutting down. With Concurrency(n) > 1, multiple input items are processed in parallel. With Ordered(), all outputs from item i are emitted before any outputs from item i+1.

When to use: Expanding records into sub-records, fetching related items per input, or flattening nested structures.

Options: Concurrency, Ordered, OnError, Buffer, Overflow, WithName, Timeout, Supervise.

// Expand each user into their orders.
orders := kitsune.FlatMap(users,
    func(ctx context.Context, u User, yield func(Order) error) error {
        orders, err := db.OrdersForUser(ctx, u.ID)
        if err != nil {
            return err
        }
        for _, o := range orders {
            if err := yield(o); err != nil {
                return err
            }
        }
        return nil
    }, kitsune.Concurrency(4))

ConcatMap

func ConcatMap[I, O any](
    p *Pipeline[I],
    fn func(context.Context, I, func(O) error) error,
    opts ...StageOption,
) *Pipeline[O]

Like FlatMap but always processes items sequentially; the next item starts only after the current item's inner pipeline has fully emitted. Output order is fully preserved. This is equivalent to FlatMap with Concurrency(1), but the intent is made explicit.

When to use: When you need strictly ordered output or when sub-streams have side effects that must not overlap.

Options: OnError, Buffer, Overflow, WithName, Timeout, Supervise.

ConcatMap is always serial. Passing Concurrency(n) with n > 1 panics at pipeline construction time; use FlatMap with Concurrency(n) if you want parallel fan-out.

// Each file is processed completely before the next starts.
lines := kitsune.ConcatMap(filePaths,
    func(ctx context.Context, path string, yield func(string) bool) error {
        f, err := os.Open(path)
        if err != nil {
            return err
        }
        defer f.Close()
        scanner := bufio.NewScanner(f)
        for scanner.Scan() {
            if err := yield(scanner.Text()); err != nil {
                return err
            }
        }
        return scanner.Err()
    })

SwitchMap

func SwitchMap[I, O any](
    p *Pipeline[I],
    fn func(context.Context, I, func(O) error) error,
    opts ...StageOption,
) *Pipeline[O]

Transforms each input into a sub-stream. When a new input arrives, the currently running sub-stream is cancelled immediately and the new one begins. Only the most-recently-started sub-stream's outputs reach downstream; older sub-streams are abandoned even if they have not finished emitting.

When to use: Typeahead search (cancel the previous request when the user types again), "latest wins" streaming; only the most recent input matters.

When not to use: When you need all outputs from all inputs, or when sub-streams have committed side effects that cannot be safely cancelled mid-flight.

Options: OnError, Buffer, Overflow, WithName, Timeout, Supervise.

// For each search query, fetch results; cancel old search when new query arrives.
results := kitsune.SwitchMap(queries,
    func(ctx context.Context, q string, yield func(Result) error) error {
        hits, err := search.Query(ctx, q)
        if err != nil {
            return err
        }
        for _, h := range hits {
            if err := yield(h); err != nil {
                return err
            }
        }
        return nil
    })

ExhaustMap

func ExhaustMap[I, O any](
    p *Pipeline[I],
    fn func(context.Context, I, func(O) error) error,
    opts ...StageOption,
) *Pipeline[O]

Transforms each input into a sub-stream, but while a sub-stream is in progress, new input items are silently dropped. Only when the current sub-stream finishes is the next item accepted. The opposite of SwitchMap, using "first wins" rather than "latest wins".

When to use: Rate-limited refresh operations where you want to ignore duplicate triggers while one is already running, e.g., a cache refresh that should not run concurrently with itself.

Options: OnError, Buffer, Overflow, WithName, Timeout, Supervise.

// Refresh cache on signal; ignore duplicate signals while refresh is in progress.
refreshed := kitsune.ExhaustMap(signals,
    func(ctx context.Context, _ Signal, yield func(CacheSnapshot) error) error {
        snap, err := rebuildCache(ctx)
        if err != nil {
            return err
        }
        return yield(snap)
    })

ExpandMap

func ExpandMap[T any](
    p *Pipeline[T],
    fn func(context.Context, T) *Pipeline[T],
    opts ...StageOption,
) *Pipeline[T]

BFS graph expansion. For each item, fn returns a child pipeline (or nil for leaf items). Items are emitted in BFS order. Use VisitedBy to detect and skip cycles.

When to use: Tree or DAG traversal where each node can produce more nodes of the same type: directory trees, dependency graphs, org charts.

Options: Buffer, WithName, MaxDepth, MaxItems, VisitedBy (for cycle detection), WithDedupSet.

Warning: unbounded by default. Without MaxDepth, MaxItems, or a downstream Take(n), ExpandMap will traverse the entire reachable graph. A graph with branching factor fan and depth d produces up to fan^d items, which can exhaust memory silently as the BFS queue grows. Always bound expansion on untrusted or potentially deep inputs.

// Crawl a directory tree.
files := kitsune.ExpandMap(
    kitsune.FromSlice([]string{"/root"}),
    func(ctx context.Context, path string) *kitsune.Pipeline[string] {
        entries, err := os.ReadDir(path)
        if err != nil {
            return nil
        }
        var children []string
        for _, e := range entries {
            if e.IsDir() {
                children = append(children, filepath.Join(path, e.Name()))
            }
        }
        return kitsune.FromSlice(children)
    },
)

Bounded expansion: cap both depth and total entries:

// Walk at most 4 levels deep and at most 10 000 entries total.
files := kitsune.ExpandMap(
    kitsune.FromSlice([]string{"/root"}),
    func(ctx context.Context, path string) *kitsune.Pipeline[string] {
        entries, err := os.ReadDir(path)
        if err != nil {
            return nil
        }
        var children []string
        for _, e := range entries {
            if e.IsDir() {
                children = append(children, filepath.Join(path, e.Name()))
            }
        }
        return kitsune.FromSlice(children)
    },
    kitsune.MaxDepth(4),
    kitsune.MaxItems(10_000),
)

When either bound is reached the stage stops enqueueing children and closes its output channel normally; no error is returned, matching the semantics of Take(n). If both options are set, whichever limit fires first wins.

Pairwise

func Pairwise[T any](p *Pipeline[T], opts ...StageOption) *Pipeline[Consecutive[T]]

Emits overlapping consecutive pairs: {item[0], item[1]}, {item[1], item[2]}, {item[2], item[3]}, …. The first item is held internally; no pair is emitted until the second item arrives. A stream of n items produces n-1 pairs.

When to use: Computing deltas between consecutive values, detecting direction changes, change detection.

Options: Buffer, WithName.

deltas := kitsune.Map(
    kitsune.Pairwise(prices),
    kitsune.LiftPure(func(c kitsune.Consecutive[float64]) float64 {
        return c.Curr - c.Prev
    }),
)

Unbatch

func Unbatch[T any](p *Pipeline[[]T], opts ...StageOption) *Pipeline[T]

Flattens a pipeline of slices into a pipeline of individual items. This is the inverse of Batch.

When to use: When you receive data in batches (bulk API response, database rows) and want to process items individually downstream.

Options: Buffer, WithName.

// API returns batches; process each item individually.
items := kitsune.Unbatch(kitsune.Map(pages, fetchPage))

Filtering & Selection

Filter

func Filter[T any](p *Pipeline[T], pred func(context.Context, T) (bool, error), opts ...StageOption) *Pipeline[T]

Emits only items for which pred returns true. Items for which pred returns false are silently dropped. Items for which pred returns an error halt the pipeline (unless OnError is set).

The method form on *Pipeline[T] accepts a simpler func(T) bool:

func (p *Pipeline[T]) Filter(fn func(T) bool, opts ...StageOption) *Pipeline[T]

Options (free function): Buffer, Overflow, WithName. OnError applies if the predicate can return errors.

// Free function with context-aware predicate:
active := kitsune.Filter(users, func(ctx context.Context, u User) (bool, error) {
    return subscriptionDB.IsActive(ctx, u.ID)
})

// Method form with simple predicate:
adults := users.Filter(func(u User) bool { return u.Age >= 18 })

IgnoreElements

func IgnoreElements[T any](p *Pipeline[T]) *Pipeline[T]

Drains p for its side effects and emits nothing downstream. The returned pipeline completes (or errors) when p completes (or errors). Any Tap, Map, or other side-effecting operators in p still run.

Also available as a method: p.IgnoreElements().

When to use: You want a pipeline to run for its side effects (writes, metrics, logging) without forwarding any items to downstream consumers.

Options: none.

// Run a Tap-instrumented pipeline for its side effects only:
kitsune.IgnoreElements(
    kitsune.Tap(events, func(_ context.Context, e Event) error {
        metrics.Record(e)
        return nil
    }),
).Run(ctx)

// Method form:
events.Tap(metrics.Record).IgnoreElements().Run(ctx)

See also: ForEach (terminal; use when you own the run), Filter (keeps matching items).

Reject

func Reject[T any](p *Pipeline[T], pred func(context.Context, T) (bool, error), opts ...StageOption) *Pipeline[T]

The inverse of Filter. Emits only items for which pred returns false (discards items where pred is true). Also available as a method with func(T) bool.

nonEmpty := strings.Reject(func(s string) bool { return s == "" })

Take

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

Emits the first n items and then stops the pipeline, signalling upstream sources to stop producing. Infinite sources like Ticker and Repeatedly stop cleanly when Take closes.

Also available as p.Take(n).

Options: none.

first10 := kitsune.Take(events, 10)

Drop

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

Discards the first n items, then forwards all subsequent items unchanged.

Also available as p.Drop(n) and p.Skip(n) (alias).

Options: none.

// Skip the header row.
data := kitsune.Drop(csvLines, 1)

TakeWhile

func TakeWhile[T any](p *Pipeline[T], pred func(T) bool) *Pipeline[T]

Emits items as long as pred returns true. As soon as pred returns false, the pipeline stops; the item that failed the predicate is not emitted. Unlike Filter, which drops individual items, TakeWhile terminates the pipeline.

Options: none.

// Stop reading when we reach a sentinel record.
data := kitsune.TakeWhile(records, func(r Record) bool { return r.Type != "EOF" })

DropWhile

func DropWhile[T any](p *Pipeline[T], pred func(T) bool) *Pipeline[T]

Discards items as long as pred returns true. The first item for which pred returns false (and all subsequent items) are forwarded.

Options: none.

// Skip header lines starting with '#'.
lines := kitsune.DropWhile(rawLines, func(s string) bool { return strings.HasPrefix(s, "#") })

TakeEvery

func TakeEvery[T any](p *Pipeline[T], n int) *Pipeline[T]

Emits every nth item starting with the first (index 0). Items at indices 1, 2, …, n-1 are dropped; the item at index n is emitted; and so on. Panics if n <= 0.

Options: none.

// Sample every 10th reading.
sampled := kitsune.TakeEvery(sensorReadings, 10)

DropEvery

func DropEvery[T any](p *Pipeline[T], n int) *Pipeline[T]

Drops every nth item (indices 0, n, 2n, …), forwarding all others. Panics if n <= 0.

Options: none.

// Drop every 5th item.
filtered := kitsune.DropEvery(events, 5)

Distinct

func Distinct[T comparable](p *Pipeline[T], opts ...StageOption) *Pipeline[T]

Emits only items that have not been seen before in the entire stream, using == equality. Keeps an in-memory set of all seen values; memory usage grows with the number of unique items.

Options: Buffer, WithName.

uniqueIDs := kitsune.Distinct(allIDs)

DistinctBy

func DistinctBy[T any, K comparable](p *Pipeline[T], keyFn func(T) K, opts ...StageOption) *Pipeline[T]

Like Distinct but uses keyFn to derive the comparison key, allowing deduplication of complex types by a single field.

Options: Buffer, WithName.

// Deduplicate events by their ID field.
unique := kitsune.DistinctBy(events, func(e Event) string { return e.ID })

Dedupe / DedupeBy

func Dedupe[T comparable](p *Pipeline[T], opts ...StageOption) *Pipeline[T]
func DedupeBy[T any, K comparable](p *Pipeline[T], keyFn func(T) K, opts ...StageOption) *Pipeline[T]

Suppresses duplicate items. By default (without DedupeWindow), deduplication is global: any item whose key was already seen anywhere in the stream is dropped. Dedupe uses == equality; DedupeBy uses a key function.

Use DedupeWindow(n) to change the deduplication scope:

• DedupeWindow(0): global deduplication (default). All previously-seen keys are remembered.
• DedupeWindow(1): consecutive deduplication only. An item is dropped only if it is identical to the immediately preceding item. This is the classic "remove adjacent duplicates" behaviour.
• DedupeWindow(n) where n > 1: sliding-window deduplication. An item is dropped if it appeared in the last n items.

Use WithDedupSet to provide a custom backend (Redis, Bloom filter) for global deduplication instead of the default in-process MemoryDedupSet.

Options: DedupeWindow, WithDedupSet, Buffer, WithName.

// Global dedup (default): drop any item seen previously in the stream.
unique := kitsune.DedupeBy(events, func(e Event) string { return e.ID })

// Consecutive-only: suppress adjacent duplicates.
changes := kitsune.DedupeBy(statusUpdates, func(s Status) string { return s.State },
    kitsune.DedupeWindow(1),
)

// Sliding window: drop any item seen in the last 50 items.
recent := kitsune.Dedupe(values, kitsune.DedupeWindow(50))

DedupSet backends

Dedupe, DedupeBy, and ExpandMap all accept a WithDedupSet(backend) option to override their default in-process deduplication store. Three built-in backends are provided:

MemoryDedupSet

func MemoryDedupSet() DedupSet

Unbounded in-memory set. Never evicts. Suitable for finite streams or when the key space is bounded. The default for all dedup operators.

BloomDedupSet

func BloomDedupSet(expectedItems int, falsePositiveRate float64) DedupSet

Bounded probabilistic set backed by a Bloom filter. Memory usage is fixed regardless of key-space size; items are never missed (zero false-negative rate), but a configured false-positive rate allows a small fraction of unseen keys to appear seen. Panics if expectedItems <= 0 or falsePositiveRate is not in (0, 1).

When to use: when the key space is unbounded but bounded memory is required and occasional false positives are acceptable (e.g. spam suppression, cache-miss avoidance).

TTLDedupSet

func TTLDedupSet(ttl time.Duration) DedupSet

In-process deduplication set that forgets keys ttl after they were last added. Memory is bounded by the set of currently non-expired keys. Eviction is lazy: expired entries are purged on the next Contains or Add call; there is no background goroutine. Re-adding an existing key refreshes its expiry (touch semantics). Panics if ttl <= 0.

When to use: deduplicating a high-volume event stream over a sliding time window (e.g. suppress duplicate webhooks received within the last 5 minutes) without unbounded memory growth. Prefer over MemoryDedupSet when keys must be forgotten; prefer over BloomDedupSet when zero false positives are required.

// Suppress duplicate webhook deliveries within a 5-minute window.
set := kitsune.TTLDedupSet(5 * time.Minute)
unique := kitsune.DedupeBy(events, func(e Event) string { return e.ID },
    kitsune.WithDedupSet(set),
)

RandomSample

func RandomSample[T any](p *Pipeline[T], rate float64, opts ...StageOption) *Pipeline[T]

Passes each item downstream with probability rate (in the range [0.0, 1.0]). Each item is evaluated independently; decisions are stateless. A rate of 1.0 passes all items; 0.0 drops all items.

When to use: Streaming telemetry sampling, load-shedding under high throughput, approximate analysis where a random subset is sufficient.

Comparison with TakeRandom: RandomSample is a streaming operator; it passes through items as they arrive and requires no buffering. TakeRandom buffers the entire stream and returns a fixed-size slice using reservoir sampling; use it when you need a precise count from a finite stream.

Options: Buffer, WithName.

// Forward roughly 5% of events to a debug trace topic.
sample := kitsune.RandomSample(events, 0.05)

Stateful Transforms

These operators inject a *Ref[S], a concurrent-safe state handle, into the stage function, enabling accumulators, counters, and running state across items.

Key / NewKey / Ref

func NewKey[T any](name string, initial T, opts ...KeyOption) Key[T]

Declares a typed, run-scoped state key. Declare keys as package-level variables. The initial value is used at the start of each runner.Run. Use StateTTL(d) to expire state after a period of inactivity.

var callCountKey = kitsune.NewKey[int]("call_count", 0)

A Ref[T] injected by MapWith provides:

• Get(ctx): read current value
• Set(ctx, value): overwrite
• Update(ctx, fn): atomic read-modify-write
• UpdateAndGet(ctx, fn): atomic read-modify-write, returns new value
• GetOrSet(ctx, fn): return existing or initialise

MapWith

func MapWith[I, O, S any](
    p *Pipeline[I],
    key Key[S],
    fn func(context.Context, *Ref[S], I) (O, error),
    opts ...StageOption,
) *Pipeline[O]

Like Map but injects a *Ref[S] carrying persistent state identified by key. At Concurrency(1) (the default), a single Ref is shared across all items in sequence; this is perfect for running totals, accumulators, or event counters. With Concurrency(n) > 1, each worker goroutine gets its own independent Ref (worker-local state).

State survives across items within a single runner.Run. Use WithStore at run time to persist state to Redis, DynamoDB, etc.

When to use: Running counters, sequence numbering, de-duplication with memory, rate tracking per pipeline run.

Options: Concurrency, OnError, Buffer, Overflow, WithName, Timeout, Supervise.

var seqKey = kitsune.NewKey[int]("seq", 0)

numbered := kitsune.MapWith(events, seqKey,
    func(ctx context.Context, ref *kitsune.Ref[int], e Event) (NumberedEvent, error) {
        n, err := ref.UpdateAndGet(ctx, func(n int) (int, error) { return n + 1, nil })
        if err != nil {
            return NumberedEvent{}, err
        }
        return NumberedEvent{Seq: n, Event: e}, nil
    },
)

FlatMapWith

func FlatMapWith[I, O, S any](
    p *Pipeline[I],
    key Key[S],
    fn func(context.Context, *Ref[S], I, func(O) error) error,
    opts ...StageOption,
) *Pipeline[O]

Like FlatMap but with a *Ref[S] for persistent state. Each input can produce zero or more outputs while reading and writing state.

Options: Concurrency, OnError, Buffer, Overflow, WithName, Timeout, Supervise.

var windowKey = kitsune.NewKey[[]Event]("window", nil)

// Emit a window every 5 events.
windows := kitsune.FlatMapWith(events, windowKey,
    func(ctx context.Context, ref *kitsune.Ref[[]Event], e Event, yield func([]Event) error) error {
        buf, _ := ref.Get(ctx)
        buf = append(buf, e)
        if len(buf) >= 5 {
            if err := yield(append([]Event(nil), buf...)); err != nil {
                return err
            }
            buf = buf[:0]
        }
        return ref.Set(ctx, buf)
    },
)