Jobpipeline Laravel Package

Getting Started

Minimal Setup

1. Install the package:

   composer require stancl/jobpipeline
   

2. Define a job pipeline in your EventServiceProvider or a service class:

   use Stancl\JobPipeline\JobPipeline;
   use App\Jobs\ProcessOrder;
   use App\Jobs\SendConfirmation;
   use App\Jobs\UpdateInventory;
   
   // Register pipeline for an event
   Event::listen(\App\Events\OrderPlaced::class, function () {
       return JobPipeline::make([
           ProcessOrder::class,
           UpdateInventory::class,
           SendConfirmation::class,
       ])
       ->send(function (\App\Events\OrderPlaced $event) {
           return $event->order; // Pass order data to jobs
       })
       ->shouldBeQueued(true)
       ->toListener();
   });
   

3. Trigger the pipeline by dispatching the event:

   event(new OrderPlaced($order));
   

First Use Case: Event-Driven Workflow

Convert a monolithic job into a pipeline for an OrderCreated event:

// Before: Single bloated job
// After: Modular pipeline
JobPipeline::make([
    ValidateOrder::class,
    ChargePayment::class,
    FulfillOrder::class,
    NotifyCustomer::class,
])
->send(fn (OrderCreated $event) => $event->order)
->shouldBeQueued('high-priority')
->toListener();

Implementation Patterns

1. Event-Driven Pipelines

• Pattern: Attach pipelines to events in EventServiceProvider or dynamically via Event::listen().
• Example:

  // Dynamic registration (e.g., in a service class)
  Event::listen(OrderPlaced::class, JobPipeline::make([
      ProcessPayment::class,
      UpdateInventory::class,
  ])->send(fn ($event) => $event->order)->toListener());
  

2. Conditional Job Injection

• Pattern: Dynamically add/remove jobs based on event data.
• Example:

  $pipeline = JobPipeline::make([
      ValidateOrder::class,
      ChargePayment::class,
  ]);
  if ($event->order->isPremium()) {
      $pipeline->add(ApplyDiscount::class);
  }
  $pipeline->send(fn ($event) => $event->order)->toListener();
  

3. Queue Management

• Pattern: Route pipelines to specific queues for priority control.
• Example:

  JobPipeline::make([...])
      ->shouldBeQueued('payments') // High-priority queue
      ->send(fn ($event) => $event->order)
      ->toListener();
  

4. Shared Context

• Pattern: Pass context (e.g., tenant ID, user ID) to all jobs via the send() closure.
• Example:

  JobPipeline::make([...])
      ->send(fn (TenantCreated $event) => [
          'tenantId' => $event->tenant->id,
          'data' => $event->data,
      ])
      ->toListener();
  

5. Early Termination

• Pattern: Cancel remaining jobs if a job returns false.
• Example:

  // In ChargePaymentJob::handle()
  if (!$payment->succeeded()) {
      return false; // Stops pipeline execution
  }
  

6. Testing Workflows

• Pattern: Mock pipelines in tests using JobPipeline::fake() (if supported) or test individual jobs.
• Example:

  public function test_order_pipeline()
  {
      $order = new Order();
      $this->fake(); // Fake queues if using Laravel 10+
  
      event(new OrderPlaced($order));
  
      // Assert jobs were dispatched
      ProcessOrder::assertDispatched();
      ChargePayment::assertDispatched();
  }
  

7. Multi-Tenancy Context

• Pattern: Use tenancy package integration to pass tenant context.
• Example:

  JobPipeline::make([...])
      ->send(fn (TenantCreated $event) => [
          'tenant' => $event->tenant,
          'central' => CentralTenant::current(), // For multi-tenancy
      ])
      ->toListener();
  

8. Error Handling

• Pattern: Wrap pipelines in try-catch or use Laravel’s HandleExceptions.
• Example:

  try {
      event(new OrderPlaced($order));
  } catch (\Throwable $e) {
      Log::error("Pipeline failed: " . $e->getMessage());
      // Notify admin or retry logic
  }
  

Gotchas and Tips

Pitfalls

1. Queue Defaults:

   • Pipelines are synchronous by default. Set JobPipeline::$shouldBeQueuedByDefault = true globally or use shouldBeQueued(true) per pipeline.
   • Fix: Explicitly configure queuing behavior to avoid unexpected sync execution.

2. Context Leakage:

   • Jobs in a pipeline share the same context. Avoid side effects (e.g., modifying shared variables) that could break subsequent jobs.
   • Fix: Use immutable data or pass fresh context per job.

3. Early Termination:

   • Returning false from a job stops the entire pipeline. Use sparingly for critical failures.
   • Fix: Split pipelines into logical groups (e.g., payment-pipeline, fulfillment-pipeline) for granular control.

4. Error Propagation:

   • Job failures do not automatically stop the pipeline unless explicitly handled (e.g., returning false).
   • Fix: Use try-catch in jobs or wrap pipelines in error handlers.

5. Queue Connections:

   • Pipelines use the default queue connection unless specified. Ensure your queue drivers (e.g., database, redis) are configured.
   • Fix: Explicitly set the queue name: shouldBeQueued('queue-name').

6. Closure Errors:

   • Errors in the send() closure (e.g., accessing undefined properties) will crash the pipeline.
   • Fix: Validate event data before passing it to the pipeline.

7. Testing Quirks:

   • Pipelines do not support fake() natively. Test by dispatching events and asserting job dispatches.
   • Fix: Use Laravel’s assertDispatched() or mock queues manually.

Debugging Tips

1. Log Pipeline Execution: Add logging in jobs to track progress:

   public function handle()
   {
       Log::info("Executing {$this->job} with data: " . json_encode($this->data));
       // Job logic
   }
   

2. Inspect Queue Jobs: Use php artisan queue:work --once to manually process jobs and debug failures.

3. Check Queue Names: Verify queue names with:

   dd(config('queue.connections'));
   

4. Validate Context: Ensure the send() closure returns the expected data structure. Use dd() to inspect:

   ->send(fn ($event) => dd($event->data)) // Debug context
   

Extension Points

1. Custom Pipeline Classes: Extend JobPipeline to add methods (e.g., withRetry(), withTimeout()):

   class CustomPipeline extends JobPipeline
   {
       public function withRetry(int $attempts): self
       {
           $this->retryAttempts = $attempts;
           return $this;
       }
   }
   

2. Dynamic Job Resolution: Override resolveJobs() to dynamically load jobs from a config or database:

   protected function resolveJobs(array $jobClasses): array
   {
       return collect($jobClasses)->map(fn ($class) => new $class())->all();
   }
   

3. Middleware for Jobs: Add middleware to jobs in the pipeline (e.g., logging, auth):

   JobPipeline::make([...])
       ->withMiddleware([
           new LogJobExecution(),
           new ValidateTenant(),
       ])
       ->send(...);
   

4. Pipeline Events: Listen for pipeline lifecycle events (e.g., JobPipelineStarting, JobPipelineCompleted) using Laravel’s Event system.

Config Quirks

1. Global Defaults: Override defaults in bootstrap/app.php or a service provider:

   \Stancl\JobPipeline\JobPipeline::$shouldBeQueuedByDefault = true;
   

2. Queue Configuration: Ensure your .env has the correct queue driver:

   QUEUE_CONNECTION=redis
   

3. Job Dependencies: If jobs require shared dependencies (e.g., a Tenant resolver), ensure they are available in the pipeline’s context.

Performance Tips

1. Batch Processing: For high-volume pipelines, use Laravel’s job:batch to process jobs in chunks: