CommandPipe.php

<?php

use SimpleBus\Message\Message;

interface CommandPipe
{
    /**
     * @param Message $command
     * @param callable $next
     */
    public function pipe(Message $command, callable $next);
}

LazyCommandPipeline.php

<?php

use SimpleBus\Message\Handler\MessageHandler;
use SimpleBus\Message\Message;

class LazyCommandPipeline implements MessageHandler
{
    /**
     * @var string
     */
    private $handlerId;

    /**
     * @var array
     */
    private $pipeIds = array();

    /**
     * @var callable
     */
    private $serviceLocator;

    /**
     * @param string $handlerId
     * @param array $pipeIds
     * @param callable $serviceLocator
     */
    public function __construct($handlerId, array $pipeIds = array(), callable $serviceLocator)
    {
        $this->handlerId = $handlerId;
        $this->pipeIds = $pipeIds;
        $this->serviceLocator = $serviceLocator;
    }

    /**
     * {@inheritdoc}
     */
    public function handle(Message $command)
    {
        call_user_func($this->callableForNext(0), $command);
    }

    private function callableForNext($index)
    {
        $serviceLocator = $this->serviceLocator;

        if (!isset($this->pipeIds[$index])) {
            $handlerId = $this->handlerId;
            return function(Message $message) use ($handlerId, $serviceLocator) {
                $serviceLocator($handlerId)->handle($message);
            };
        }

        $pipeId = $this->pipeIds[$index];

        return function(Message $message) use ($pipeId, $serviceLocator, $index) {
            $pipe = $serviceLocator($pipeId);
            $pipe->pipe($message, $this->callableForNext($index + 1));
        };
    }
}