Stream Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require windwalker/stream ^4.0
   

   Ensure your composer.json includes "windwalker/stream": "^4.0" under require.

2. First Use Case: Import the Stream class and create a basic stream pipeline:

   use Windwalker\Stream\Stream;
   
   $stream = new Stream();
   $stream->pipe(function ($data) {
       return strtoupper($data);
   })->pipe(function ($data) {
       return $data . '!';
   });
   
   $result = $stream->send('hello');
   // Output: 'HELLO!'
   

3. Key Entry Points:

   • Stream class: Core class for building pipelines.
   • Stream::pipe(): Add processing steps to the pipeline.
   • Stream::send(): Execute the pipeline with input data.
   • Documentation: Windwalker Stream Docs (refer here for advanced features like map, filter, or reduce).

Implementation Patterns

Core Workflows

1. Data Transformation Pipeline

Use pipe() to chain transformations:

$stream = new Stream();
$stream
    ->pipe(fn($data) => trim($data))          // Trim whitespace
    ->pipe(fn($data) => explode(',', $data))  // Split by comma
    ->pipe(fn($data) => array_map('strtolower', $data)); // Lowercase each item

$result = $stream->send('  Apple, Banana,  Cherry  ');
// Output: ['apple', 'banana', 'cherry']

2. Filtering Data

Combine filter() with closures:

$stream = new Stream([1, 2, 3, 4, 5]);
$stream->filter(fn($item) => $item % 2 === 0);
$result = $stream->all(); // [2, 4]

3. Aggregation with reduce()

Sum values in a collection:

$stream = new Stream([10, 20, 30]);
$sum = $stream->reduce(fn($carry, $item) => $carry + $item, 0);
// $sum = 60

4. Laravel Integration

Use streams in service providers or controllers:

// app/Providers/AppServiceProvider.php
public function boot()
{
    $stream = new Stream();
    $stream->pipe(fn($data) => User::where('name', 'like', "%{$data}%")->get());

    $users = $stream->send('John');
}

5. Async Processing (with Promises)

For async operations (if supported in future versions), leverage then():

$stream = new Stream();
$stream->pipe(fn($data) => \DB::table('logs')->where('data', $data)->get())
       ->then(fn($result) => $this->processResults($result));

Integration Tips

With Laravel Collections

Convert streams to collections or vice versa:

$collection = collect([1, 2, 3]);
$stream = new Stream($collection->toArray());
$stream->map(fn($item) => $item * 2);
$result = $stream->toCollection(); // Collects [2, 4, 6]

Custom Stream Classes

Extend Stream for domain-specific logic:

class UserStream extends Stream
{
    public function __construct()
    {
        $this->pipe(fn($data) => User::find($data));
    }
}

$userStream = new UserStream();
$user = $userStream->send(1); // Fetches user with ID 1

Error Handling

Wrap streams in try-catch blocks:

try {
    $result = $stream->send($input);
} catch (\Exception $e) {
    Log::error("Stream error: " . $e->getMessage());
    return response()->json(['error' => 'Processing failed'], 500);
}

Gotchas and Tips

Pitfalls

1. Immutable Data Flow

• Streams are immutable by default. Each pipe() or filter() returns a new stream:

  $stream1 = new Stream([1, 2, 3]);
  $stream2 = $stream1->filter(fn($item) => $item > 1);
  // $stream1 remains unchanged; $stream2 contains [2, 3]
  

• Fix: Reassign streams after modifications or use tap() for side effects:

  $stream->tap(fn($s) => $s->log('Processing...'));
  

2. Memory Leaks with Large Data

• Avoid loading entire datasets into memory. Use chunk() or cursor() for large datasets:

  $stream = new Stream();
  $stream->cursor(fn() => DB::table('large_table')->cursor());
  

3. Closure Scope Issues

• Closures in pipe() may not have access to outer variables. Use use or bind variables:

  $userId = 1;
  $stream->pipe(fn($data) => User::find($userId)->posts()->where('data', $data)->get());
  // Better: Pass $userId as a parameter or use a class method.
  

4. Undefined Methods

• The package may not implement all array/collection methods (e.g., pluck(), sort()). Fall back to Laravel collections or implement custom methods:

  $stream->map(fn($item) => $item->toArray())->then(fn($data) => collect($data)->pluck('id'));
  

Debugging Tips

1. Log Stream State

Use tap() to inspect intermediate results:

$stream->tap(fn($data) => Log::debug('Current data:', $data));

2. Check for Silent Failures

Ensure all pipe() closures return valid data. Invalid returns (e.g., null) may break subsequent steps.

3. Validate Input/Output

Add assertions for critical pipelines:

$stream->pipe(fn($data) => assert(is_array($data), 'Expected array input'))
       ->pipe(fn($data) => array_filter($data));

Extension Points

1. Custom Stream Drivers

Extend Stream to support custom data sources (e.g., API streams, database cursors):

class ApiStream extends Stream
{
    public function __construct(string $endpoint)
    {
        $this->pipe(fn() => Http::get($endpoint)->json());
    }
}

2. Middleware for Streams

Create reusable pipeline logic:

$stream->pipe(new class {
    public function __invoke($data) {
        return str_replace([' ', '-'], '_', $data);
    }
});

3. Event Dispatching

Trigger events at specific pipeline stages:

$stream->pipe(fn($data) => event(new StreamProcessed($data)));

4. Configuration

Centralize stream configurations (e.g., default pipes) in a config file:

// config/stream.php
'default_pipes' => [
    'sanitize' => fn($data) => filter_var($data, FILTER_SANITIZE_STRING),
];

Then load them dynamically:

$stream = new Stream(config('stream.default_pipes'));

Laravel-Specific Quirks

1. Service Container Binding

Bind the Stream class for dependency injection:

$this->app->bind(Stream::class, fn() => new Stream());

Then inject it into controllers:

public function __construct(private Stream $stream) {}

2. Queueable Streams

For long-running streams, dispatch to queues:

dispatch(new class extends Job {
    public function handle() {
        $stream = new Stream();
        $stream->pipe(fn($data) => $this->processData($data));
        $stream->send($data);
    }
});

3. View Composition

Use streams to transform data before passing to Blade:

$stream = new Stream($products);
$stream->map(fn($product) => [
    'name' => $product->name,
    'price' => '$' . $product->price,
]);
return view('products', ['products' => $stream->all()]);