DependencyInjection/ONGRConnectionsExtension.php
protected function createPipelines(ContainerBuilder $container, array $config)
    {
        foreach ($config['pipelines'] as $pipelineName => $pipelineConfig) {
            if (!isset($pipelineConfig['shop'])) {
                $pipelineConfig['shop'] = $container->getParameter('ongr_connections.active_shop');
            }
            $serviceConfig = $this->prepareServiceConfigs($container, $pipelineConfig, $pipelineName);
            $this->createServices(
                $container,
                $pipelineConfig['provide_sources'],
                $serviceConfig,
                "data_sync.{$pipelineName}.source",
                'onSource'
            );
            $this->createServices(
                $container,
                $pipelineConfig['provide_consumers'],
                $serviceConfig,
                "data_sync.{$pipelineName}.consume",
                'onConsume'
            );
            foreach ($pipelineConfig['types'] as $type => $typeConfig) {
                $typeServiceConfig = $this->prepareTypeServiceConfigs($serviceConfig, $typeConfig, $type);
                $serviceList = $this->getServiceList($pipelineName, $type);
                foreach ($serviceList as $name => $service) {
                    $this->createServices(
                        $container,
                        array_merge($pipelineConfig[$name], $typeConfig[$name]),
                        $typeServiceConfig,
                        $service['tag'],
                        $service['method']
                    );
                }
            }
        }
    }
protected function createPipelines(ContainerBuilder $container, array $config)
    {
        foreach ($config['pipelines'] as $pipelineName => $pipelineConfig) {
            if (!isset($pipelineConfig['shop'])) {
                $pipelineConfig['shop'] = $container->getParameter('ongr_connections.active_shop');
            }
            $serviceConfig = $this->prepareServiceConfigs($container, $pipelineConfig, $pipelineName);
            $this->createServices(
                $container,
                $pipelineConfig['provide_sources'],
                $serviceConfig,
                "data_sync.{$pipelineName}.source",
                'onSource'
            );
            $this->createServices(
                $container,
                $pipelineConfig['provide_consumers'],
                $serviceConfig,
                "data_sync.{$pipelineName}.consume",
                'onConsume'
            );
            foreach ($pipelineConfig['types'] as $type => $typeConfig) {
                $typeServiceConfig = $this->prepareTypeServiceConfigs($serviceConfig, $typeConfig, $type);
                $serviceList = $this->getServiceList($pipelineName, $type);
                foreach ($serviceList as $name => $service) {
                    $this->createServices(
                        $container,
                        array_merge($pipelineConfig[$name], $typeConfig[$name]),
                        $typeServiceConfig,
                        $service['tag'],
                        $service['method']
                    );
                }
            }
        }
    }
Creates sync, import and provide pipeline services.

@param ContainerBuilder $container
@param array            $config