Flow Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require darkwood/flow
   

   Requires PHP 8.5+.

2. First Flow Creation:

   use Flow\FlowFactory;
   use Flow\Ip;
   
   $flow = (new FlowFactory())->create(function() {
       yield fn($input) => $input + 1; // Simple transformation
   });
   
   $ip = new Ip(5);
   $flow($ip);
   $flow->await(); // Outputs: 6
   

3. Key Classes:

   • FlowFactory: Creates flow instances
   • Ip (Input Pointer): Holds current data and context
   • Flow: The executable flow pipeline

First Use Case: Data Pipeline

Transform sequential data processing (e.g., API responses → parsed → enriched → stored):

$flow = (new FlowFactory())->create(function() {
    yield fn($apiResponse) => json_decode($apiResponse, true);
    yield fn($parsedData) => array_map('strtoupper', $parsedData);
    yield fn($uppercaseData) => ['processed_at' => now(), 'data' => $uppercaseData];
});

$ip = new Ip(file_get_contents('api/endpoint'));
$flow($ip);
$flow->await(); // Returns enriched data

Implementation Patterns

Core Workflow: Functional Pipeline

1. Define Jobs:

   $flow = (new FlowFactory())->create(function() {
       // Job 1: Input → Intermediate
       yield fn($input) => transform($input);
   
       // Job 2: Intermediate → Output
       yield fn($intermediate) => finalize($intermediate);
   });
   

2. Execute:

   $ip = new Ip($initialData);
   $flow($ip);
   $result = $ip->data; // Final output
   $flow->await(); // Trigger async execution
   

Async Integration Patterns

1. Driver Selection:

   // Configure default driver (e.g., Amp)
   FlowFactory::setDefaultDriver(new AmpDriver());
   

2. Driver-Specific Jobs:

   use Flow\Jobs\AsyncJob;
   
   $flow = (new FlowFactory())->create(function() {
       yield new AsyncJob(fn() => sleep(1)); // Non-blocking sleep
   });
   

Y-Combinator for Recursion

use Flow\Jobs\YJob;

// Recursive flow (e.g., tree traversal)
$Ywrap = static function(callable $func, callable $wrapper) {
    return new YJob(static fn($recurse) => $wrapper($recurse));
};

$flow = (new FlowFactory())->create(function() use ($Ywrap) {
    yield $Ywrap(
        fn($recurse) => fn($node) => $node->children ? array_map($recurse, $node->children) : $node->value,
        fn($recurse) => fn($node) => $recurse($node)
    );
});

Laravel Integration

1. Service Provider:

   public function register()
   {
       $this->app->singleton(FlowFactory::class, function($app) {
           return (new FlowFactory())->setDefaultDriver(new AmpDriver());
       });
   }
   

2. Command Example:

   use Illuminate\Console\Command;
   use Flow\FlowFactory;
   
   class ProcessDataCommand extends Command
   {
       protected $signature = 'data:process';
       protected $description = 'Process data pipeline';
   
       public function handle(FlowFactory $flowFactory)
       {
           $flow = $flowFactory->create(function() {
               yield fn($data) => $this->transform($data);
               yield fn($transformed) => $this->store($transformed);
           });
   
           $flow(new Ip($this->getInputData()));
           $flow->await();
       }
   }
   

Gotchas and Tips

Pitfalls

1. Driver Compatibility:

   • Gotcha: Not all drivers support PHP 8.5+ (e.g., older Spatie Async versions).
   • Fix: Check composer.json constraints or use parallel driver for CPU-bound tasks.

2. State Management:

   • Gotcha: Ip (Input Pointer) is immutable—modify data via returned objects.
   • Fix: Avoid direct property manipulation; return new instances:

     yield fn($ip) => new Ip($ip->data + 1); // Correct
     $ip->data++; // Anti-pattern (silently fails)
     

3. Async Deadlocks:

   • Gotcha: Blocking calls (e.g., file_get_contents) in async flows halt execution.
   • Fix: Use async drivers (Amp/React) for I/O:

     yield new AsyncJob(fn() => Amp\async(function() {
         return file_get_contents('url');
     }));
     

4. Y-Combinator Stack Overflow:

   • Gotcha: Deep recursion may exhaust stack limits.
   • Fix: Limit recursion depth or use iterative alternatives.

Debugging Tips

1. Flow Tracing:

   $flow = (new FlowFactory())->create(function() {
       yield fn($data) => tap($data, fn($d) => logger()->debug("Processing: {$d}"));
   });
   

2. Driver-Specific Logs:

   • Enable Amp/React logging via their respective config (e.g., AMP_LOG_LEVEL=debug).

3. Ip Inspection:

   $flow->on('step', fn($ip) => logger()->debug("Current data:", $ip->data));
   

Extension Points

1. Custom Drivers: Implement Flow\DriverInterface for new async backends:

   class MyDriver implements DriverInterface {
       public function run(callable $job): void { /* ... */ }
       public function await(): void { /* ... */ }
   }
   

2. Job Decorators: Wrap jobs for cross-cutting concerns (e.g., logging, retries):

   class LoggingJob implements JobInterface {
       public function __construct(private JobInterface $job, private Logger $logger) {}
       public function run(Ip $ip): Ip { /* Log + delegate */ }
   }
   

3. Flow Middleware:

   $flow = (new FlowFactory())->create(function() {
       yield new MiddlewareJob(fn($next) => fn($ip) => $next($ip)->tap(fn($ip) => $this->postProcess($ip)));
   });
   

Performance Quirks

1. Fiber Driver:

   • Tip: Use for CPU-bound tasks (faster than threads).
   • Warning: Not all PHP extensions are fiber-safe (e.g., some database drivers).

2. Parallel Driver:

   • Tip: Ideal for multi-core workloads but adds complexity.
   • Warning: Shared memory requires careful data isolation.

3. Amp Driver:

   • Tip: Best for I/O-bound tasks (HTTP, DB).
   • Warning: Avoid mixing with blocking code.

Laravel-Specific

1. Queue Integration:

   use Flow\Jobs\QueueJob;
   
   $flow = (new FlowFactory())->create(function() {
       yield new QueueJob('process-data', $data); // Dispatch to queue
   });
   

2. Event Dispatching:

   $flow = (new FlowFactory())->create(function() {
       yield fn($data) => event(new DataProcessed($data));
   });
   

3. Service Container Binding:

   $this->app->bind(FlowFactory::class, function($app) {
       $factory = new FlowFactory();
       $factory->setDefaultDriver($app->make(AmpDriver::class));
       return $factory;
   });