code: 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;
      }
docs: 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