Weave Code
Pipeline Laravel Package
amphp/pipeline
Fiber-safe concurrent iterators and collection operators for AMPHP. Build pipelines from iterables, map/filter/merge, and consume results from multiple fibers safely using ConcurrentIterator (foreach or manual continue/getValue/getPosition). Requires PHP 8.1+.
Concurrent iterators and pipeline operations.
Frequently asked questions about Pipeline
- Can I use amphp/pipeline in Laravel without AMPHP or Swoole?
- No, this package requires a fiber-compatible runtime like AMPHP, Swoole, or ReactPHP. For traditional Laravel (CLI/FPM), use it with worker pools like RoadRunner or Spiral Framework. If you’re not using fibers, consider synchronous alternatives like Laravel Collections or queue-based parallelism via spatie/async.
- How do I process database results asynchronously with Laravel and amphp/pipeline?
- Replace `DB::get()` with `DB::cursor()` to stream rows, then wrap it in a pipeline: `Pipeline::fromIterable(DB::cursor())->concurrent(5)->map(fn($row) => $row->process())->forEach(fn($result) => Log::info($result));`. This avoids loading all rows into memory at once.
- Will amphp/pipeline work with Laravel Horizon for queue jobs?
- Yes, but Horizon workers must run in a fiber-compatible environment (e.g., RoadRunner). Use `Pipeline::fromIterable(Job::all())->concurrent(10)->map(fn($job) => $job->handle())` to process jobs in parallel within a worker. For non-fiber setups, use spatie/async or Horizon’s built-in parallelism.
- What’s the difference between `ConcurrentIterator` and Laravel Collections?
- Laravel Collections are synchronous and single-threaded, while `ConcurrentIterator` enables fiber-safe, parallel processing across multiple fibers. Use Collections for simple transformations; use `amphp/pipeline` for async workflows like batch API calls or high-volume database exports where concurrency is critical.
- How do I handle errors in a pipeline if a fiber fails?
- Wrap pipeline operations in `try/catch` blocks. Exceptions propagate to the fiber consuming the iterator. For resilience, implement circuit-breaker patterns (e.g., retry failed jobs) or use `Pipeline::buffer()` to limit concurrent operations and prevent cascading failures.
- Can I use amphp/pipeline for real-time event processing in Laravel?
- Yes, transform event payloads asynchronously before dispatching: `Pipeline::fromIterable($events)->map(fn($event) => $event->sanitize())->forEach(fn($event) => event($event));`. This works with Laravel’s event system if your app runs in a fiber-compatible environment like RoadRunner.
- What’s the performance impact of using concurrent iterators vs. synchronous loops?
- Concurrent iterators introduce context-switching overhead, so benchmark against synchronous alternatives (e.g., `Collection::chunk()`). Use pipelines for I/O-bound tasks (e.g., API calls, DB queries) where parallelism outweighs the cost. For CPU-bound work, synchronous loops may still be faster.
- How do I limit memory usage with large datasets in amphp/pipeline?
- Use `Pipeline::buffer($size)` to cap concurrent operations, preventing memory spikes. For example, `Pipeline::fromIterable(DB::cursor())->buffer(100)->map(...)` ensures no more than 100 rows are processed concurrently. This is critical for rate-limited APIs or large exports.
- Are there alternatives to amphp/pipeline for async Laravel workflows?
- For non-fiber setups, use Laravel’s built-in queues (Horizon) or packages like `spatie/async` for parallel job processing. If you need fiber support, consider `reactphp/promise` or `swoole/async`, but `amphp/pipeline` is the most integrated solution for AMPHP’s concurrency model.
- How do I migrate an existing Laravel app to use amphp/pipeline?
- Start with a non-critical async workflow (e.g., image resizing). Replace `foreach ($items)` with `Pipeline::fromIterable($items)->foreach(fn($item) => ...)`. Gradually adopt it for queue jobs or database cursors. Use RoadRunner or Spiral Framework to manage fibers in production.
Weaver
How can I help you explore Laravel packages today?