Pipeline Laravel Package

Getting Started

Install. composer require sanmai/pipeline — latest release needs PHP 8.2+, so it drops into Laravel 10/11 projects without conflict. There are no service providers, no config file, no facade. It's a pure library: one class (Pipeline\Standard) plus five namespaced functions.

Where to look first. In this order:

1. src/functions.php — the five entry points (take(), map(), fromArray(), fromValues(), zip()). This is the whole surface you'll type 90% of the time.
2. src/Standard.php — skim the method list, not the bodies. Note which methods are terminal (toList(), toAssoc(), count(), each(), fold(), first(), last(), min(), max(), finalVariance()) versus which just append a lazy stage.
3. The Read the Docs site (php-functional-pipeline.readthedocs.io), especially the cookbook and best-practices pages — the README is a reference, the docs are the tutorial.

First use case: a streaming export from Eloquent. The most natural day-one win in a Laravel app is replacing Model::all() + array_map + array_filter with a lazily-pulled cursor:

use function Pipeline\take;

$rows = take(User::query()->where('active', true)->cursor())
    ->cast(fn (User $u) => ['id' => $u->id, 'email' => $u->email])
    ->select(fn (array $r) => $r['email'] !== null)   // strict: drops null/false only
    ->toList();

The query streams row-by-row; nothing is hydrated into a collection; the array is only materialised at toList(). That single pattern — take(<source>)->cast(...)->select(...)->toList() — is the mental model for everything else in this assessment.

Implementation Patterns

Ingest with take(), adapt with cast(), trim with select(). take() accepts arrays, generators, iterators, and variadics merged in succession — so you can concatenate two Eloquent builders' cursors, a SplFileObject, and a range() in one call. Use cast() for strict 1:1 transforms (preserves keys, no generator interpretation) and map() only when a callback must yield zero-or-many.

One-to-many expansion without nested loops. map() is SelectMany: whatever a callback yields is spliced into the stream.

take(Order::query()->cursor())
    ->map(function (Order $order) {
        foreach ($order->lines as $line) {
            yield $line;              // flattened into the stream
        }
    })
    ->cast(fn (OrderLine $l) => $l->sku)
    ->toList();

The equivalent with flatten() + unpack() is fine too, but the generator form keeps memory flat. Pair this with ->with('lines') on the query; see the unbuffered-query gotcha below.

Batching for writes. chunk() is the idiomatic replacement for ->chunk(500) on a builder when you don't want to re-query:

take(ImportRow::cursor())
    ->cast(fn ($r) => $this->normalise($r))
    ->chunk(500)
    ->each(fn (array $batch) => DB::table('imports')->insert($batch));

chunkBy() does the same with variable sizes (a size-0 chunk yields an empty array), useful for grouping by a run-length condition such as consecutive log entries from the same tenant.

Side effects without mutation. tap() observes each element and passes it through untouched — the right place for Log::debug, counters, or accumulating a checksum. each() is the terminal version. Both receive ($value, $key); the library catches ArgumentCountError once and permanently downgrades arity-sensitive callables (like printf(...)) to single-argument form, so single-arg closures are safe and cheaper.

Fan-in with zip(). When two sources must be paired positionally — a CSV of skus and an array of prices — zip() transposes them and unpack() destructures:

take($skus)
    ->zip($prices)
    ->unpack(fn (string $sku, int $price) => [$sku => $price])
    ->toAssoc();

Summary objects with fold(). Build a multi-field aggregate in one pass instead of three separate ->sum() queries:

$stats = take($events)->fold(
    ['count' => 0, 'bytes' => 0],
    function (array $carry, Event $e) {
        $carry['count']++;
        $carry['bytes'] += $e->size;
        return $carry;
    }
);

Statistics without buffering. runningVariance() and finalVariance() implement Welford's online algorithm — sample mean, variance, standard deviation in O(1) memory. Use these instead of collect($rows)->avg() + ->stddev() over a big stream.

Early exit. Because iteration is pull-based, a plain foreach with break stops all upstream work immediately. For "give me the first 20 that match" this is strictly better than filtering a full collection:

foreach (take($hugeGenerator)->select($predicate) as $row) {
    $page[] = $row;
    if (count($page) >= 20) break;   // upstream stops here
}

peek(20) gets you the first N as a pipeline (destructively — see gotchas); prepend() restores them.

Package boundary discipline. Pipelines are mutable and single-use, so never return one from a public method. Treat the pipeline as an implementation detail of a method whose contract is array, a generator, or a DTO. Wrap recurring pipelines in small invokable classes or a readonly service, and always finish with toList() / toAssoc() / fold() at the boundary.

Gotchas and Tips

Illuminate\Pipeline\Pipeline is a totally different class. Laravel's Pipeline is the middleware/->through()->then() request pipeline. If a file imports both, alias explicitly: use Pipeline\Standard as Stream;. This is the single most common source of confusion for Laravel devs picking this package up.

Nothing runs until you consume. Every stage is a deferred generator stage. A pipeline assigned to a variable and never iterated is a no-op, and a bug that "does nothing" is almost always a missing terminal call. Consume with foreach, toList(), toAssoc(), each(), or fold().

DB::cursor() + lazy callbacks = unbuffered-query landmine. MySQL/Postgres refuse further queries on the same connection while an unbuffered result set is open. If your map()/tap() closures touch the database (or trigger lazy-loaded relations), you'll get "Cannot execute queries while other unbuffered queries are active" or a quiet N+1 storm. Eager-load with ->with(...), or use ->lazy(500) instead of ->cursor() so the pipeline sits between chunk queries.

Don't use iterator_to_array(). Keys are preserved on a best-effort basis, so duplicate keys silently collapse to the last value. Use toList() (drops keys, keeps every value) or toAssoc() (keeps keys). This bites hardest on map() stages that yield repeated keys.

Pipelines are not rewindable. Second consumption of a generator-backed pipeline raises PHP's "Cannot traverse an already closed generator." The library itself defines no exceptions, so this is a language error surfacing through a toList() you expected to be idempotent. If you need to pause and resume, call ->cursor() — it returns a NoRewindIterator variant that auto-advances past the current element on resume (like a database cursor) rather than repeating it. If you need to iterate strictly once with accidental-reset protection, cursor() again. If you need an actual Iterator, wrap: new \IteratorIterator($pipeline).

Mutability is viral. Every stage method returns $this, so $other = $pipeline->select(...) also mutated $pipeline. Convenient when injecting stages from a helper; dangerous if two code paths share an instance. Never store a pipeline on a singleton or a shared service.

Array inputs execute eagerly. If the pipeline is seeded with an array (or a map() callback that returns instead of yields), the library stores an array internally and eagerly shortcuts everything with array_map/array_filter. When in doubt, yield. stream() forces lazy, non-array paths.

filter() is not select(). select() is strict — it drops only null and false. filter() is the array_filter alias that drops every falsy value, including 0 and ''. Filtering numeric data with filter() will delete your zeroes.

peek() is destructive. It eagerly consumes the first N items and leaves the remainder in the pipeline; the consumed items are gone unless you prepend() them back.

count() is terminal. Counting exhausts the pipeline. For a count alongside processing, use runningCount($ref) with a by-reference counter instead.

An unprimed pipeline fails silently. new Standard() with no iterator and no seed generator makes most methods no-ops (internal empty() gate). If a result is unexpectedly [], check the seed first — there is no exception to guide you.

Debuggability is manual by design. No exceptions means bad input yields empty output, not an error. Put tap(fn ($v, $k) => Log::debug('stage', compact('v', 'k'))) between stages, and ->peek(5)->toList() at the head to eyeball the source. reservoir(n) samples a bounded random subset of a huge stream for inspection without materialising it.

Testing. Assertions must come after a terminal call — a pipeline object alone proves nothing. In PHPUnit, build the pipeline inside the test, call toList(), and assert on the array. Since the class also implements Countable, assertCount() works but consumes the instance; order your assertions or rebuild.

Extension points. Subclass Pipeline\Standard to add domain stages (->onlyActive()), keeping return $this. The library ships generic type annotations, so on Larastan/PHPStan take(Model::query()->cursor()) infers through cast()/select() reasonably well — annotate closures' parameters to get the most out of it.