.github/workflows/php.yml

@@ -0,0 +1,20 @@
+name: PHP Composer
+
+on:
+  push:
+    branches: [ master 4.2 4.1 3.1 1.6 ]
+  pull_request:
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+
+    - name: Install dependencies
+      run: php -r "copy('https://cs.symfony.com/download/php-cs-fixer-v3.phar', 'php-cs-fixer.phar');"
+
+    - name: Run php-cs-fixer 
+      run: php php-cs-fixer.phar fix --dry-run --config=.php_cs.php --cache-file=.php_cs.cache --verbose --show-progress=dots --diff --allow-risky=yes

.gitignore

@@ -1,2 +1,3 @@
 *.phar
 *.cache
+.idea

.php_cs.php

@@ -1,17 +1,18 @@
 <?php
 
-$finder = PhpCsFixer\Finder::create()
+$finder = \PhpCsFixer\Finder::create()
     ->in(__DIR__);
 
-return PhpCsFixer\Config::create()
+return (new \PhpCsFixer\Config())
     ->setFinder($finder)
     ->setRules(
         [
             // generic PSRs
             '@PSR1' => true,
             '@PSR2' => true,
-            'psr0' => true,
-            'psr4' => true,
+            '@PSR12' => true,
+            '@PSR12:risky' => true,
+            'psr_autoloading' => true,
 
             // imports
             'ordered_imports' => true,
@@ -39,11 +40,13 @@
             'escape_implicit_backslashes' => false,
 
             // PHP
-            '@PHP71Migration' => true,
-            '@PHP71Migration:risky' => true,
+            '@PHP73Migration' => true,
+            '@PHP74Migration' => true,
+            '@PHP74Migration:risky' => true,
 
+            'use_arrow_functions' => false,
+            'get_class_to_class_keyword' => false,
             'void_return' => false,
-            'visibility_required' => false,
             'list_syntax' => ['syntax' => 'long'],
             'declare_strict_types' => false,
 
@@ -64,11 +67,11 @@
             '@Symfony:risky' => true,
             'phpdoc_types_order' => false,
             'phpdoc_separation' => false,
-            'phpdoc_inline_tag' => false,
+            'visibility_required' => ['elements' => ['property', 'method']],
+            'types_spaces' => false,
             'native_function_invocation' => false,
             'concat_space' => ['spacing' => 'one'],
             'single_space_after_construct' => false,
-            'trailing_comma_in_multiline_array' => false,
             'self_accessor' => false,
             'yoda_style' => false,
             'phpdoc_summary' => false,
@@ -87,6 +90,7 @@
             'ternary_operator_spaces' => false,
             'phpdoc_no_useless_inheritdoc' => false,
             'class_definition' => false,
+            'string_length_to_empty' => false,
         ]
     )
     ->setRiskyAllowed(true);

Async/ImportProductProcessor.php

@@ -3,14 +3,15 @@
 namespace Oro\Bundle\AkeneoBundle\Async;
 
 use Doctrine\ORM\EntityManagerInterface;
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
 use Oro\Bundle\IntegrationBundle\Authentication\Token\IntegrationTokenAwareTrait;
 use Oro\Bundle\IntegrationBundle\Entity\Channel as Integration;
+use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges;
 use Oro\Bundle\IntegrationBundle\Provider\SyncProcessorRegistry;
-use Oro\Bundle\SecurityBundle\Authentication\Token\UsernamePasswordOrganizationToken;
+use Oro\Bundle\MessageQueueBundle\Entity\Job;
 use Oro\Component\MessageQueue\Client\TopicSubscriberInterface;
 use Oro\Component\MessageQueue\Consumption\MessageProcessorInterface;
-use Oro\Component\MessageQueue\Job\Job;
 use Oro\Component\MessageQueue\Job\JobRunner;
 use Oro\Component\MessageQueue\Transport\MessageInterface;
 use Oro\Component\MessageQueue\Transport\SessionInterface;
@@ -20,6 +21,7 @@
 
 class ImportProductProcessor implements MessageProcessorInterface, TopicSubscriberInterface
 {
+    use CacheProviderTrait;
     use IntegrationTokenAwareTrait;
 
     /** @var DoctrineHelper */
@@ -91,37 +93,42 @@ public function process(MessageInterface $message, SessionInterface $session)
             return self::REJECT;
         }
 
-        $this->setTemporaryUserIntegrationToken($integration);
+        $this->setTemporaryIntegrationToken($integration);
 
         $result = $this->jobRunner->runDelayed(
             $body['jobId'],
             function (JobRunner $jobRunner, Job $child) use ($integration, $body) {
                 $this->doctrineHelper->refreshIncludingUnitializedRelations($integration);
                 $processor = $this->syncProcessorRegistry->getProcessorForIntegration($integration);
+
+                $em = $this->doctrineHelper->getEntityManager(FieldsChanges::class);
+                /** @var FieldsChanges $fieldsChanges */
+                $fieldsChanges = $em
+                    ->getRepository(FieldsChanges::class)
+                    ->findOneBy(['entityId' => $child->getId(), 'entityClass' => Job::class]);
+
+                if (!$fieldsChanges) {
+                    $this->logger->error(
+                        sprintf('Source data from Akeneo not found for job: %s', $child->getId())
+                    );
+
+                    return false;
+                }
+
+                $this->cacheProvider->save('akeneo', $fieldsChanges->getChangedFields());
+
                 $status = $processor->process(
                     $integration,
                     $body['connector'] ?? null,
                     $body['connector_parameters'] ?? []
                 );
 
+                $em->clear(FieldsChanges::class);
+
                 return $status;
             }
         );
 
-        $this->setTemporaryIntegrationToken($integration);
-
         return $result ? self::ACK : self::REJECT;
     }
-
-    private function setTemporaryUserIntegrationToken(Integration $integration)
-    {
-        $token = new UsernamePasswordOrganizationToken(
-            $integration->getDefaultUserOwner(),
-            $integration->getDefaultUserOwner()->getUsername(),
-            'main',
-            $integration->getOrganization()
-        );
-        $token->setAttribute('owner_description', 'Integration: ' . $integration->getName());
-        $this->tokenStorage->setToken($token);
-    }
 }

Async/SyncIntegrationProcessor.php

@@ -9,7 +9,6 @@
 use Oro\Bundle\IntegrationBundle\Provider\LoggerStrategyAwareInterface;
 use Oro\Bundle\IntegrationBundle\Provider\SyncProcessorRegistry;
 use Oro\Bundle\MessageQueueBundle\Entity\Job;
-use Oro\Bundle\SecurityBundle\Authentication\Token\UsernamePasswordOrganizationToken;
 use Oro\Component\MessageQueue\Client\TopicSubscriberInterface;
 use Oro\Component\MessageQueue\Consumption\MessageProcessorInterface;
 use Oro\Component\MessageQueue\Job\JobRunner;
@@ -106,7 +105,7 @@ public function process(MessageInterface $message, SessionInterface $session)
 
         $em->getConnection()->getConfiguration()->setSQLLogger(null);
 
-        $this->setTemporaryUserIntegrationToken($integration);
+        $this->setTemporaryIntegrationToken($integration);
         $integration->getTransport()->getSettingsBag()->set('page_size', $body['transport_batch_size']);
 
         $result = $this->jobRunner->runUnique(
@@ -128,20 +127,6 @@ function (JobRunner $jobRunner, Job $job) use ($integration, $body) {
             }
         );
 
-        $this->setTemporaryIntegrationToken($integration);
-
         return $result ? self::ACK : self::REJECT;
     }
-
-    private function setTemporaryUserIntegrationToken(Integration $integration)
-    {
-        $token = new UsernamePasswordOrganizationToken(
-            $integration->getDefaultUserOwner(),
-            $integration->getDefaultUserOwner()->getUsername(),
-            'main',
-            $integration->getOrganization()
-        );
-        $token->setAttribute('owner_description', 'Integration: ' . $integration->getName());
-        $this->tokenStorage->setToken($token);
-    }
 }

Command/CleanupCommand.php

@@ -0,0 +1,104 @@
+<?php
+
+namespace Oro\Bundle\AkeneoBundle\Command;
+
+use Oro\Bundle\BatchBundle\ORM\Query\BufferedIdentityQueryResultIterator;
+use Oro\Bundle\CronBundle\Command\CronCommandInterface;
+use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
+use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges;
+use Oro\Bundle\MessageQueueBundle\Entity\Job;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+/**
+ * Clears old records from oro_integration_fields_changes table before repack.
+ */
+class CleanupCommand extends Command implements CronCommandInterface
+{
+    /** @var string */
+    protected static $defaultName = 'oro:cron:akeneo:cleanup';
+
+    /** @var DoctrineHelper */
+    private $doctrineHelper;
+
+    public function __construct(DoctrineHelper $doctrineHelper)
+    {
+        $this->doctrineHelper = $doctrineHelper;
+        parent::__construct();
+    }
+
+    public function isActive()
+    {
+        return true;
+    }
+
+    public function getDefaultDefinition()
+    {
+        return '0 2 * * 6';
+    }
+
+    public function configure()
+    {
+        $this
+            ->setDescription('Clears old records from oro_integration_fields_changes table.')
+            ->setHelp(
+                <<<'HELP'
+                    The <info>%command.name%</info> command clears fields changes for complete job records
+                    from <comment>oro_integration_fields_changes</comment> table.
+
+                      <info>php %command.full_name%</info>
+                    HELP
+            );
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $output->writeln(sprintf(
+            '<comment>Number of fields changes that has been deleted:</comment> %d',
+            $this->deleteRecords()
+        ));
+
+        $output->writeln('<info>Fields changes cleanup complete</info>');
+    }
+
+    private function deleteRecords(): int
+    {
+        $qb = $this->doctrineHelper
+            ->getEntityManagerForClass(FieldsChanges::class)
+            ->getRepository(FieldsChanges::class)
+            ->createQueryBuilder('fc');
+
+        $qb
+            ->delete(FieldsChanges::class, 'fc')
+            ->where($qb->expr()->eq('fc.entityClass', ':class'))
+            ->setParameter('class', Job::class)
+            ->andWhere($qb->expr()->in('fc.entityId', ':ids'));
+
+        $jqb = $this->doctrineHelper
+            ->getEntityManagerForClass(Job::class)
+            ->getRepository(Job::class)
+            ->createQueryBuilder('j');
+
+        $jqb
+            ->select('j.id')
+            ->where($jqb->expr()->in('j.status', ':statuses'))
+            ->setParameter('statuses', [Job::STATUS_SUCCESS, Job::STATUS_CANCELLED, Job::STATUS_FAILED, Job::STATUS_STALE])
+            ->orderBy($jqb->expr()->desc('j.id'));
+
+        $iterator = new BufferedIdentityQueryResultIterator($jqb->getQuery());
+
+        $result = 0;
+        $iterator->setPageLoadedCallback(function (array $rows) use ($qb, &$result): array {
+            $ids = array_column($rows, 'id');
+
+            $result = $result + $qb->setParameter('ids', $ids)->getQuery()->execute();
+
+            return $ids;
+        });
+
+        iterator_to_array($iterator);
+
+        return $result;
+    }
+}

Controller/ValidateConnectionController.php

@@ -52,6 +52,7 @@ public function validateConnectionAction(Request $request, Channel $channel = nu
             $akeneoSettings->setPassword($akeneoSettingsEntity->getPassword());
         }
 
+        /** @var \Oro\Bundle\CurrencyBundle\Provider\CurrencyListProviderInterface */
         $currencyConfig = $this->container->get('oro_currency.config.currency');
 
         $akeneoChannelNames = [];
@@ -93,7 +94,7 @@ public function validateConnectionAction(Request $request, Channel $channel = nu
                 'akeneoLocales' => $akeneoLocales,
                 'success' => $success,
                 'message' => $this->get('translator')->trans($message),
-                'currencyList' => $currencyConfig->getCurrencies(),
+                'currencyList' => $currencyConfig->getCurrencyList(),
             ]
         );
     }

DependencyInjection/OroAkeneoExtension.php

@@ -22,6 +22,7 @@ class OroAkeneoExtension extends Extension
     public function load(array $configs, ContainerBuilder $container)
     {
         $loader = new Loader\YamlFileLoader($container, new FileLocator(__DIR__ . '/../Resources/config'));
+        $loader->load('commands.yml');
         $loader->load('integration.yml');
         $loader->load('importexport.yml');
         $loader->load('services.yml');

Entity/AkeneoSettings.php

@@ -89,6 +89,12 @@ class AkeneoSettings extends Transport
      * @ORM\Column(name="akeneo_product_filter", type="text", nullable=true)
      */
     protected $productFilter;
+    /**
+     * @var string
+     *
+     * @ORM\Column(name="akeneo_conf_product_filter", type="text", nullable=true)
+     */
+    protected $configurableProductFilter;
     /**
      * @var string
      *
@@ -275,6 +281,26 @@ public function setProductFilter($productFilter)
         return $this;
     }
 
+    /**
+     * @return string
+     */
+    public function getConfigurableProductFilter()
+    {
+        return $this->configurableProductFilter;
+    }
+
+    /**
+     * @param string $configurableProductFilter
+     *
+     * @return self
+     */
+    public function setConfigurableProductFilter($configurableProductFilter)
+    {
+        $this->configurableProductFilter = $configurableProductFilter;
+
+        return $this;
+    }
+
     /**
      * @return ParameterBag
      */