Pipeline Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require amphp/pipeline
   

   Requires PHP 8.1+.

2. First Use Case: Transform a simple array into a concurrent pipeline and process it:

   use Amp\Pipeline\Pipeline;
   
   $pipeline = Pipeline::fromIterable([1, 2, 3, 4, 5])
       ->map(fn($n) => $n * 2)
       ->filter(fn($n) => $n % 3 === 0);
   
   foreach ($pipeline as $value) {
       echo $value . "\n"; // Output: 6
   }
   

3. Key Entry Points:

   • Pipeline::fromIterable(): Convert arrays/generators to pipelines.
   • Pipeline::generate(): Create pipelines from closures.
   • Queue: For async producer-consumer patterns.

Implementation Patterns

1. Data Processing Pipelines

Pattern: Chain operations like map, filter, tap for async data flows.

$pipeline = Pipeline::fromIterable(range(1, 100))
    ->concurrent(10) // Process 10 items concurrently
    ->tap(fn() => Amp\delay(0.1)) // Simulate I/O
    ->map(fn($n) => $n * 2)
    ->filter(fn($n) => $n % 5 === 0)
    ->reduce(fn($carry, $n) => $carry + $n, 0);

Workflow:

• Use concurrent(N) for parallelism (e.g., API calls, DB queries).
• unordered() for out-of-order results (e.g., logging).
• buffer(N) to control back-pressure (e.g., throttling).

2. Async Producer-Consumer

Pattern: Use Queue for fiber-safe async communication.

$queue = new Queue();
Amp\async(function() use ($queue) {
    foreach (range(1, 5) as $n) {
        $queue->push($n); // Blocks until consumed
    }
    $queue->complete();
});

foreach ($queue->iterate() as $value) {
    echo $value . "\n"; // Consumes asynchronously
}

Integration Tips:

• Pair with Amp\async for fire-and-forget producers.
• Use pushAsync() for non-blocking pushes (returns a Future).

3. Merging Streams

Pattern: Combine multiple pipelines with merge().

$pipeline1 = Pipeline::fromIterable([1, 2, 3]);
$pipeline2 = Pipeline::fromIterable(['a', 'b']);
$merged = Pipeline::merge($pipeline1, $pipeline2);
foreach ($merged as $value) {
    echo $value . "\n"; // Output: 1, 2, 3, a, b
}

Use Case: Aggregate logs from multiple sources or batch API responses.

4. Error Handling

Pattern: Use catch() or try-catch with Queue::error().

$queue = new Queue();
Amp\async(function() use ($queue) {
    try {
        foreach (range(1, 3) as $n) {
            $queue->push($n);
        }
    } catch (Exception $e) {
        $queue->error($e); // Propagates to consumer
    }
    $queue->complete();
});

Gotchas and Tips

Pitfalls

1. Forgetting complete():

   • Omitting $queue->complete() will hang consumers indefinitely.
   • Fix: Always call complete() or error() after pushing all values.

2. Back-Pressure Mismanagement:

   • Without buffer(N), producers block on every push.
   • Fix: Use buffer(5) to allow 5 items in flight.

3. Concurrent Iterator State:

   • getPosition() may not reflect sequential order across fibers.
   • Fix: Use foreach for simplicity; avoid manual continue() unless necessary.

4. DisposedException:

   • Thrown if a consumer is disposed before all values are pushed.
   • Fix: Catch DisposedException in producers to handle early termination.

Debugging Tips

• Check Iterator State:

  if ($iterator->isComplete()) {
      echo "Iterator finished\n";
  }
  

• Log Pipeline Steps:

  ->tap(fn($value) => error_log("Processing: $value"))
  

• Use Amp\delay() to simulate slow operations and test back-pressure.

Extension Points

1. Custom Operators: Create reusable pipeline methods:

   $pipeline->customOp(fn($pipeline) => $pipeline->map(fn($n) => $n * 2));
   

2. Integrate with Laravel:

   • Use Queue for async job processing (e.g., Laravel Queues).
   • Example:

     $queue = new Queue();
     Amp\async(function() use ($queue) {
         foreach (Job::chunk(10) as $job) {
             $queue->push($job->handle());
         }
         $queue->complete();
     });
     

3. Performance Tuning:

   • Adjust concurrent(N) based on system resources (e.g., N=5 for I/O-bound tasks).
   • Use unordered() for non-sequential tasks (e.g., parallel HTTP requests).

Config Quirks

• Default Buffer Size: 0 (blocks on every push). Set buffer(N) to optimize throughput.
• Generator Memory: Avoid large generators in fromIterable(); use Queue for streaming.

```markdown
### Laravel-Specific Notes
1. **Async Job Queues**:
   Replace Laravel’s `dispatch()` with `Queue` for fiber-based workers:
   ```php
   $queue = new Queue();
   Amp\async(function() use ($queue) {
       foreach (Job::all() as $job) {
           $queue->push($job->fire());
       }
       $queue->complete();
   });

2. Event Listeners: Use Pipeline to process events concurrently:

   event(new UserRegistered($user))
       ->then(fn() => Pipeline::fromIterable([$user])
           ->concurrent(3)
           ->tap(fn($user) => $this->notify($user))
           ->tap(fn($user) => $this->log($user)));
   

3. Database Queries: Batch queries with concurrent():

   $pipeline = Pipeline::fromIterable(User::cursor())
       ->concurrent(20)
       ->map(fn($user) => $user->load('posts'))
       ->toArray();