Pipeline Laravel Package

Getting Started

Install (usually already present — amphp/http-client, amphp/socket, amphp/mysql, and amphp/redis all pull it in transitively):

composer require amphp/pipeline

Requirements: PHP 8.1+, Revolt event loop, and an async runtime context (Amp\async(), or the top-level main fiber, which revolts runs the loop for).

Where to look first — the package surface is tiny, read these three in order:

1. src/Pipeline.php — the final class with all operators (map, filter, tap, concurrent, buffer, take, merge, concat, delay, forEach, reduce) plus factories fromIterable(), generate().
2. src/ConcurrentIterator.php — the fiber-safe iterator contract (continue(), getValue(), getPosition(), isComplete(), dispose()).
3. src/Queue.php — the producer/consumer bridge with push(), pushAsync(), complete(), error().

First real use case — you have a foreach that awaits an I/O call per item and it's painfully serial. Convert it to a pipeline and fan out:

use Amp\Pipeline\Pipeline;

$bodies = Pipeline::fromIterable($urls)   // lazy: nothing runs yet
    ->concurrent(10)                       // run up to 10 downstream ops at once
    ->map(fn (string $url) => $httpClient->request(new Request($url))->getBody())
    ->toArray();

That's the 80% story: fromIterable(...)->concurrent(N)->map(...) and consume. Everything else is a variation on it. If you only ever use Queue, you're still getting the core value: turning callback/event-driven producers into something you can foreach while keeping back-pressure.

Implementation Patterns

Lazy sources: pass a closure, not an array

Pipeline::fromIterable() accepts an iterable or a Closure returning an iterable. Passing a generator function (not an instantiated generator) is the idiom for expensive/lazy production:

$pages = Pipeline::fromIterable(function () use ($client): \Generator {
    $cursor = null;
    do {
        $page = $client->listPage($cursor);
        yield from $page->items;
        $cursor = $page->nextCursor;
    } while ($cursor !== null);
});

Nothing is produced until the pipeline is consumed — this is what makes ->take(5) cheap against an infinite source.

concurrent(N) is the workhorse

concurrent(n) forks up to n fibers for everything downstream of it. The canonical shape:

Pipeline::generate(fn (): int => ++$i)
    ->take(100)
    ->concurrent(8)
    ->tap(fn () => Amp\delay(0.1))      // simulate/perform I/O per item
    ->map(fn (int $v) => transform($v))
    ->filter(fn ($v) => $v !== null)
    ->forEach(fn ($v) => $sink->write($v));

Note the operator ordering matters: concurrent() only overlaps the operators after it. Put cheap CPU work before it, I/O after.

Reordering results from an unordered pipeline

concurrent() is inherently unordered (the ordered variant was removed in beta.2 because it broke with filtering operators). If you need stable output, carry an index through and sort at the end:

$results = Pipeline::fromIterable($items)
    ->map(fn ($item, int $pos) => [$pos, $item])  // keep the position
    ->concurrent(10)
    ->map(fn (array [$pos, $item]) => [$pos, awaitWork($item)])
    ->toArray();

ksort($results); // or usort by the captured position

Alternatively, use tap() for the side-effecting I/O and keep the ordering semantics in your own array keyed by position.

Queue: adapting push-style APIs into iterators

This is where the package earns its keep. Any producer that isn't a loop (websocket frames, stream chunks, an SDK that takes a callback) can be exposed as a ConcurrentIterator:

use Amp\Pipeline\Queue;

$queue = new Queue();          // optional: new Queue($bufferSize) to allow slack
$socket->onMessage(fn (string $data) => $queue->push($data));

try {
    foreach ($queue->iterate() as $chunk) {
        $parser->feed($chunk);
        if ($parser->done()) {
            break;             // disposes the iterator, see Gotchas
        }
    }
} finally {
    $queue->complete();        // ALWAYS: the consumer hangs forever otherwise
    $socket->close();
}

• push() suspends the producer until the value is consumed — that's back-pressure. Use it from an Amp\async() fiber.
• pushAsync() returns a Future immediately — use it when you want fire-and-forget but still want to observe DisposedException or await all pending pushes (Future\awaitAll()).
• error($throwable) poisons the iterator — the consumer's continue()/foreach throws.

Consuming a ConcurrentIterator from multiple fibers

The whole point of the interface: one iterator, many consumers. Get the iterator once and share the object:

$iterator = Pipeline::fromIterable($jobs)->getIterator();

$workers = [];
for ($i = 0; $i < 8; ++$i) {
    $workers[] = Amp\async(function () use ($iterator, $i): void {
        foreach ($iterator as $position => $job) {   // each value goes to exactly one fiber
            process($job);
        }
    });
}
Amp\Future\await($workers);

Each value is delivered to exactly one consumer — this is a work-queue, not a broadcast. Also note $position is not sequential within a single fiber.

Early exit and cancellation

break out of a foreach disposes the iterator, which is what unwinds producers and relieves back-pressure. For timeouts, pass a Cancellation to continue():

$iterator = $pipeline->getIterator();
while ($iterator->continue(new TimeoutCancellation(5))) {
    handle($iterator->getValue());
}

Wrap iteration in try/finally and call $iterator->dispose() explicitly if you need deterministic cleanup before returning from an outer scope.

Fan-in with merge() / concat()

• Pipeline::merge([$a, $b]) (1.1+) emits whenever any source emits — good for multiplexing sockets or combining independent job streams.
• Pipeline::concat([$a, $b]) drains in order — good for "retry page N, then continue".

Integration notes

• Pipelines compose naturally with amphp/http-client request() loops, amphp/mysql execute() loops, and file read() loops. If you find yourself writing foreach { await(...) } against any of these, it's a pipeline candidate.
• Use tap() (not map()) for logging/metrics/timing so you don't change the value or the cardinality.
• Use buffer(n) (1.2+) to decouple a fast producer from a bursty consumer; raise Pipeline throughput without changing concurrent().
• Test pipelines with Amp\PHPUnit\AsyncTestCase; assertions on ordering need async-aware tests, and synchronous PHPUnit will deadlock the first time an operator awaits.

Gotchas and Tips

Forgetting complete() is the #1 hang. A Queue that is never completed or errored leaves every consumer's continue() suspended forever. Put complete() in a finally, not at the end of the happy path. (Destructors won't save you — the behavior was deliberately removed in beta.7 because destruct order is random.)

DisposedException from push() is normal, not an error. When the consumer breaks, calls take(), or the iterator is garbage-collected, pending and future push() calls throw DisposedException. In a producer fiber, catch it (or ignore the future from pushAsync()) and stop producing gracefully. The same instance is created once per queue, so identity comparison is safe.

Ordering is not preserved by concurrent(). Do not assume foreach over a concurrent() pipeline yields in source order — it doesn't, and the ordered operator was removed precisely because it couldn't be made correct alongside filtering operators. Carry an index if you need order (see above).

concurrent(N) only affects downstream operators. ->concurrent(10)->map(fn () => cpuBound()) gives you 10 fibers doing CPU work — which in PHP is slower than serial because there's no parallelism, only interleaving. Use concurrent() around suspension points (I/O, Amp\delay(), Future::await()), not around computation.

Exceptions in map/filter/tap fail the whole pipeline. The exception surfaces from the consumer's continue() (or foreach body). It does not skip the item. If you want per-item fault tolerance, catch inside the callback and return a tagged result:

->map(fn ($item) => try { return [null, work($item)]; }
                  catch (\Throwable $e) { return [$e, null]; })

Then handle failures when you iterate. Otherwise one bad URL kills a 1,000-item batch.

Back-pressure defaults to "none" in some places and "always" in others. Queue::push() awaits consumption of every value; the no-arg constructor gives you zero buffering. If you're wrapping a high-frequency event source (websocket ticks, stream reads), pass a $bufferSize to Queue or add ->buffer(n) in the pipeline, otherwise you'll serialize the producer to the slowest consumer.

Unbounded memory from pushAsync(). Ignoring the returned futures means nothing ever waits, so the queue grows as fast as the producer can push. Either await them periodically (Future\awaitAll($pending)) or use push().

Stay on 1.2.x. Several bugs in this package manifest as deadlocks, not exceptions, so version drift is expensive:

• 1.1.0 — take() previously didn't complete until an extra value was emitted (a hang against finite/infinite sources).
• 1.2.1 — race between Cancellation on continue() and a simultaneous Queue::complete() threw "Must call suspend() before calling throw()".
• 1.2.2/1.2.3 — disposed Queues with unconsumed values left back-pressure futures unresolved (fiber leak).

Manual iteration has an ordering contract. You must call continue() and have it return before getValue()/getPosition(). Don't call getValue() twice expecting fresh data — the value is stable within the fiber until the next continue(). And with multiple fibers, getPosition() is a global index, not a per-fiber counter.

Don't foreach the same Pipeline twice and expect both loops to see everything. Values are distributed among consumers. If you need broadcast semantics, materialize with toArray() or duplicate the source (call fromIterable() again with a fresh generator closure).

Arrays vs generators change laziness, not results. fromIterable([1,2,3]) is eager; fromIterable(fn () => gen()) is lazy. Mixing them up usually shows up as "my take(1) still ran 10,000 queries."

Debugging a stalled pipeline. Wrap suspicious stages with ->tap(fn ($v) => printf("stage X: %s\n", json_encode($v))) to find where values stop. Check $iterator->isComplete() — true means the source is done (successfully or errored) and nothing is pending. If a fiber is stuck with no output, suspect a missing complete(), a consumer that exited without disposing, or a Queue whose producer fiber was never started.

Extension points. Implement ConcurrentIterator yourself if your source has semantics Queue can't express (e.g., a resumable cursor with its own buffering); it's a small interface, and anything implementing it works in foreach. For everything else, compose from Pipeline operators rather than hand-rolling a custom iterator — the built-in ones already handle disposal, back-pressure relief, and cancellation correctly.