pipeline

The Pipeline Pattern is an architectural pattern that encapsulates sequential processes. It functions like a production line where a payload (or subject) is passed through a series of stages. Each stage can perform specific operations such as acting on, manipulating, decorating, or replacing the payload.

Use this package when you find yourself manually passing the results of one function into another to complete a series of tasks on a single subject. This implementation allows you to mix and match individual operations and pipelines to create flexible execution chains.

Because PipelineInterface extends StageInterface, a Pipeline instance can be passed into the pipe() method of another pipeline. This allows you to build complex execution patterns by nesting smaller, specialized pipelines within larger ones.

$processApiRequest = (new Pipeline)
    ->pipe(new ExecuteHttpRequest)
    ->pipe(new ParseJsonResponse);
    
$pipeline = (new Pipeline)
    ->pipe(new ConvertToPsr7Request)
    ->pipe($processApiRequest)
    ->pipe(new ConvertToResponseDto);
    
$pipeline->process(new DeleteBlogPost($postId));

Because PipelineInterface extends StageInterface, a pipeline can be used as a stage within another pipeline. This allows for highly composable, nested execution patterns.

$processApiRequest = (new Pipeline)
    ->pipe(new ExecuteHttpRequest)
    ->pipe(new ParseJsonResponse);
    
$pipeline = (new Pipeline)
    ->pipe(new ConvertToPsr7Request)
    ->pipe($processApiRequest) // The nested pipeline acts as a single stage
    ->pipe(new ConvertToResponseDto);
    
$pipeline->process(new DeleteBlogPost($postId));

$processApiRequest = (new Pipeline)
    ->pipe(new ExecuteHttpRequest) // 2
    ->pipe(new ParseJsonResponse); // 3
    
$pipeline = (new Pipeline)
    ->pipe(new ConvertToPsr7Request) // 1
    ->pipe($processApiRequest) // (2,3)
    ->pipe(new ConvertToResponseDto); // 4 
    
$pipeline->process(new DeleteBlogPost($postId));

To create a pipeline, instantiate League\Pipeline\Pipeline and use the pipe() method to add stages. A stage can be any object that implements the __invoke magic method. Use the process() method to execute the pipeline with an initial payload. Each stage receives the output of the previous stage as its input.

use League\Pipeline\Pipeline;

class TimesTwoStage
{
    public function __invoke($payload)
    {
        return $payload * 2;
    }
}

class AddOneStage
{
    public function __invoke($payload)
    {
        return $payload + 1;
    }
}

$pipeline = (new Pipeline)
    ->pipe(new TimesTwoStage)
    ->pipe(new AddOneStage);

// Returns 21
$pipeline->process(10);

For more complex logic, you can use classes as stages. To ensure the correct method signature, implement the League\Pipeline\StageInterface. The class must implement the __invoke method to process the payload.

use League\Pipeline\Pipeline;
use League\Pipeline\StageInterface;

class TimesTwoStage implements StageInterface
{
    public function __invoke($payload)
    {
        return $payload * 2;
    }
}

class AddOneStage implements StageInterface
{
    public function __invoke($payload)
    {
        return $payload + 1;
    }
}

$pipeline = (new Pipeline)
    ->pipe(new TimesTwoStage)
    ->pipe(new AddOneStage);

// Returns 21
$pipeline->process(10);

When upgrading to version 0.2.0, any class implementing StageInterface must use the magic __invoke method instead of a custom process() method to handle the payload.

// After 0.2.0, implement __invoke instead of process
class MyStage implements StageInterface
{
    public function __invoke($payload)
    {
        return $payload;
     }
}

The recommended way to install the league/pipeline package is through Composer.

composer require league/pipeline

The package is transparent regarding exceptions: it does not catch or silence any errors. Exceptions can occur either inside a specific stage or during the process() call. You should handle exceptions either within the stage's logic or by wrapping the process() call in a try-catch block.

$pipeline = (new Pipeline)->pipe(function () {
    throw new LogicException();
});
    
try {
    $pipeline->process($payload);
} catch(LogicException $e) {
    // Handle the exception.
}

The pipeline package is transparent regarding exceptions; it does not catch or silence errors. You should handle exceptions either within a specific stage's logic or by wrapping the process() call in a try-catch block.

$pipeline = (new Pipeline)
    ->pipe(function () {
        throw new LogicException();
    });
    
try {
    $pipeline->process($payload);
} catch(LogicException $e) {
    // Handle the exception.
}

In version 0.2.0, you no longer need to wrap callables using CallableStage::forCallable(). You can pass any callable (such as an anonymous function) directly to the pipe() method.

// After 0.2.0, pass the callable directly
$pipeline->pipe(function ($payload) {
    return $payload;
})->process($payload);

Any callable can be used as a stage in a pipeline. This includes anonymous functions (closures) and any invokable object.

$pipeline = (new Pipeline)->pipe(function ($payload) {
    return $payload * 10;
});