Pipeline/PipelineFactory.php
public function create($pipelineName, $listeners = [])
    {
        $listeners = array_merge(
            [
                'sources' => [],
                'modifiers' => [],
                'consumers' => [],
            ],
            $listeners
        );

        $className = $this->getClassName();
        /** @var PipelineInterface $pipeline */
        $pipeline = new $className($pipelineName);

        if (!$pipeline instanceof Pipeline) {
            throw new \InvalidArgumentException('Pipeline class\' name must implement PipelineInterface');
        }

        $pipeline->setProgressBar($this->getProgressBar());

        $dispatcher = $this->getDispatcher();
        $pipeline->setDispatcher($dispatcher);

        foreach ($listeners['consumers'] as &$listener) {
            if ($listener === self::CONSUMER_RETURN) {
                $listener = function (ItemPipelineEvent $event) {
                    $event->setOutput($event->getItem());
                };
            }
        }

        $registerListener = function ($key, $suffix) use ($listeners, $dispatcher, $pipeline) {
            foreach ($listeners[$key] as $listener) {
                $dispatcher->addListener(
                    $pipeline->getEventName($suffix),
                    $listener
                );
            }
        };
        $registerListener('sources', Pipeline::EVENT_SUFFIX_SOURCE);
        $registerListener('modifiers', Pipeline::EVENT_SUFFIX_MODIFY);
        $registerListener('consumers', Pipeline::EVENT_SUFFIX_CONSUME);

        return $pipeline;
    }
public function create($pipelineName, $listeners = [])
    {
        $listeners = array_merge(
            [
                'sources' => [],
                'modifiers' => [],
                'consumers' => [],
            ],
            $listeners
        );

        $className = $this->getClassName();
        /** @var PipelineInterface $pipeline */
        $pipeline = new $className($pipelineName);

        if (!$pipeline instanceof Pipeline) {
            throw new \InvalidArgumentException('Pipeline class\' name must implement PipelineInterface');
        }

        $pipeline->setProgressBar($this->getProgressBar());

        $dispatcher = $this->getDispatcher();
        $pipeline->setDispatcher($dispatcher);

        foreach ($listeners['consumers'] as &$listener) {
            if ($listener === self::CONSUMER_RETURN) {
                $listener = function (ItemPipelineEvent $event) {
                    $event->setOutput($event->getItem());
                };
            }
        }

        $registerListener = function ($key, $suffix) use ($listeners, $dispatcher, $pipeline) {
            foreach ($listeners[$key] as $listener) {
                $dispatcher->addListener(
                    $pipeline->getEventName($suffix),
                    $listener
                );
            }
        };
        $registerListener('sources', Pipeline::EVENT_SUFFIX_SOURCE);
        $registerListener('modifiers', Pipeline::EVENT_SUFFIX_MODIFY);
        $registerListener('consumers', Pipeline::EVENT_SUFFIX_CONSUME);

        return $pipeline;
    }
Creates pipeline and registers first listeners in dispatcher.

Available listeners:
- $listeners['sources']: pass source listeners.
- $listeners['modifiers']: pass modifier listeners.
- $listeners['consumers']: pass consumer listeners.

@param string $pipelineName
@param array  $listeners

@return \ONGR\ConnectionsBundle\Pipeline\PipelineInterface
@throws \InvalidArgumentException