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;
    }
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
Creates pipeline and registers first listeners in dispatcher .