Pipeline Laravel Package

Product Decisions This Supports

Build vs. buy on concurrency primitives. amphp/pipeline answers a question most teams answer badly by default: do we hand-roll bounded-concurrency workers, buffering, and back-pressure, or adopt a library? The hard parts here are not the map/filter/concurrent operators — they are the fiber-safety semantics of a ConcurrentIterator consumable from multiple fibers, and the back-pressure contract on Queue::push() (producer suspends until the consumer takes the value) plus DisposedException signaling when a consumer walks away. The 1.2.x patch history — a fiber race condition fixed in 1.2.1, back-pressure not cleared on dispose fixed in 1.2.2, a regression fixed in 1.2.3 — is the strongest evidence in the repo: this is subtle territory where in-house implementations silently leak memory or hang. Buy.

Roadmap scoping for throughput-bound features. This supports "yes, we can commit to that" on features whose cost is dominated by fan-out I/O rather than CPU: bulk sync of N records against a third-party API, per-row enrichment in an import, streaming exports while a downstream consumer drains, multi-source aggregation via merge(), and bounded buffering via buffer() (added in 1.2.0) to decouple a bursty producer from a slow sink without unbounded memory. Each of these converts from "add more queue workers" (linear cost) to "raise concurrent(n) within one worker" (sub-linear cost, bounded memory). That is a direct roadmap lever.

Architecture/infra decisions. Whether to standardize on AMPHP as our async substrate at all. This package is a leaf building block in that ecosystem — it is what amphp/http-client, amphp/socket, and friends are built on. Adopting it is a smaller decision than adopting the ecosystem, but it only pays off inside it; standalone in a Laravel app it is an answer to a question we probably aren't asking.

Explicitly not supported: Laravel-native concurrency ergonomics. Http::pool(), Bus::batch(), and LazyCollection already cover the common cases for teams that will not run a Revolt event loop. Choosing this package is choosing a different concurrency model for a specific subsystem — not a general productivity upgrade.

When to Consider This Package

Adopt when:

• We already run the AMPHP/Revolt event loop in the target service (existing amphp/http-client, amphp/websocket, amphp/redis usage), so this is a zero-marginal-cost dependency.
• The workload is bounded-memory fan-out I/O — hundreds to millions of items, each requiring a network call, where holding results in an array is not viable and unbounded buffering is a production incident waiting to happen.
• Multiple fibers must consume the same stream safely (this is the specific problem the package exists to solve; a plain Iterator corrupts under concurrent consumption).
• PHP 8.1+ is already a given, and we can keep the work isolated in a CLI command, queue worker, or dedicated service process rather than mixing it into request-handling code.

Look elsewhere when:

• The app is conventional synchronous Laravel. Reach for Http::pool(), Bus::batch(), or chunked job dispatch. Lower cognitive load, better observability, no event-loop hazards.
• Team is standardizing on Swoole/Octane coroutines — mixing Revolt fibers with Swoole coroutines in one process is an incident generator.
• We need vendor support, an SLA, or a large hiring pool. 83 stars and 0 recorded dependents (almost certainly a data artifact, but the signal stands) means a small maintainer base and no commercial backing. Bus factor is a real line item.
• Blocking I/O is unavoidable in the hot path. Any file_get_contents, sleep, or synchronous DB driver inside a fiber stalls the entire loop — the failure mode is a global slowdown, not a local error, which is hard to attribute during an incident.

Adoption preconditions I would require: pin the version exactly, wrap it behind our own Pipeline-shaped adapter interface so it is swappable, restrict usage to one isolated process type, and instrument queue depth / in-flight counts before the first production deploy.

How to Pitch It (Stakeholders)

To executives: This is a small, MIT-licensed, dependency-free building block that lets one worker process handle I/O-bound work — API fan-out, record sync, streaming ingest — at bounded concurrency instead of one-item-at-a-time. The business case is cost and predictability: throughput scales by raising a concurrency number rather than by adding workers, and memory stays flat because producers are throttled by consumers instead of buffering into RAM. Cost of adoption is low (no license, no infrastructure change) provided it stays inside the service where we already run AMPHP. The honest counterweight: it is a specialist library with a thin maintainer base, so treat it as a version-pinned internal dependency, not a platform bet.

To engineering: Fibers plus a fiber-safe ConcurrentIterator plus real back-pressure, with the sharp edges already filed down — the 1.2.x releases are fixes for exactly the races and disposal leaks you would hit writing this yourself in month three. concurrent(n), buffer(), merge(), take(), tap(), map(), filter() compose declaratively. Non-negotiables: PHP 8.1+, Revolt loop, no blocking calls inside fibers, and every Queue must be complete()d or error()d — an uncompleted queue suspends its consumer forever, which is our most likely first production bug. Handle DisposedException in producers as normal termination, not an error. Isolate into a dedicated worker; wrap behind our own interface; instrument before shipping.