.github/workflows/php.yml

@@ -0,0 +1,20 @@
+name: PHP Composer
+
+on:
+  push:
+    branches: [ master 5.0 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

.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,14 @@
             'escape_implicit_backslashes' => false,
 
             // PHP
-            '@PHP71Migration' => true,
-            '@PHP71Migration:risky' => true,
+            '@PHP74Migration' => true,
+            '@PHP74Migration:risky' => true,
+            '@PHP80Migration' => true,
+            '@PHP80Migration: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 +68,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 +91,7 @@
             'ternary_operator_spaces' => false,
             'phpdoc_no_useless_inheritdoc' => false,
             'class_definition' => false,
+            'string_length_to_empty' => false,
         ]
     )
     ->setRiskyAllowed(true);

Async/ImportProductProcessor.php

@@ -3,13 +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\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;
@@ -19,6 +21,7 @@
 
 class ImportProductProcessor implements MessageProcessorInterface, TopicSubscriberInterface
 {
+    use CacheProviderTrait;
     use IntegrationTokenAwareTrait;
 
     /** @var DoctrineHelper */
@@ -97,12 +100,31 @@ public function process(MessageInterface $message, SessionInterface $session)
             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;
             }
         );

Async/SyncIntegrationProcessor.php

@@ -4,6 +4,7 @@
 
 use Doctrine\ORM\EntityManagerInterface;
 use Doctrine\Persistence\ManagerRegistry;
+use Oro\Bundle\AkeneoBundle\Integration\AkeneoChannel;
 use Oro\Bundle\IntegrationBundle\Async\Topics;
 use Oro\Bundle\IntegrationBundle\Authentication\Token\IntegrationTokenAwareTrait;
 use Oro\Bundle\IntegrationBundle\Entity\Channel as Integration;
@@ -117,7 +118,9 @@ function (JobRunner $jobRunner, Job $job) use ($integration, $body) {
                     $processor->getLoggerStrategy()->setLogger($this->logger);
                 }
                 $connectorParameters = $body['connector_parameters'];
-                $connectorParameters['rootJobId'] = $job->getRootJob()->getId();
+                if ($integration->getType() === AkeneoChannel::TYPE) {
+                    $connectorParameters['rootJobId'] = $job->getRootJob()->getId();
+                }
 
                 return $processor->process(
                     $integration,

Command/CleanupCommand.php

@@ -0,0 +1,106 @@
+<?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>');
+
+        return 0;
+    }
+
+    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;
+    }
+}

Config/ConfigManager.php

@@ -32,4 +32,32 @@ public function hasChanges(): bool
 
         return false;
     }
+
+    protected function getDiff($values, $originalValues)
+    {
+        $diff = [];
+        if (empty($originalValues)) {
+            foreach ($values as $code => $value) {
+                $diff[$code] = [null, $value];
+            }
+        } else {
+            foreach ($originalValues as $code => $originalValue) {
+                if (array_key_exists($code, $values)) {
+                    $value = $values[$code];
+                    if ($originalValue != $value) {
+                        $diff[$code] = [$originalValue, $value];
+                    }
+                } else {
+                    $diff[$code] = [$originalValue, null];
+                }
+            }
+            foreach ($values as $code => $value) {
+                if (!array_key_exists($code, $originalValues)) {
+                    $diff[$code] = [null, $value];
+                }
+            }
+        }
+
+        return $diff;
+    }
 }

Controller/ValidateConnectionController.php

@@ -7,6 +7,7 @@
 use Oro\Bundle\IntegrationBundle\Entity\Channel;
 use Oro\Bundle\IntegrationBundle\Form\Type\ChannelType;
 use Oro\Bundle\SecurityBundle\Annotation\Acl;
+use Psr\Http\Client\ClientExceptionInterface;
 use Sensio\Bundle\FrameworkExtraBundle\Configuration\ParamConverter;
 use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
 use Symfony\Component\HttpFoundation\JsonResponse;
@@ -52,6 +53,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 = [];
@@ -78,7 +80,7 @@ public function validateConnectionAction(Request $request, Channel $channel = nu
                     $akeneoCurrencies = $transport->getMergedCurrencies();
                     $akeneoLocales = $transport->getLocales();
             }
-        } catch (ExceptionInterface $e) {
+        } catch (ClientExceptionInterface | ExceptionInterface $e) {
             $success = false;
             $message = $e->getMessage();
         } catch (\Exception $e) {
@@ -93,7 +95,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/AkeneoLocale.php

@@ -53,9 +53,6 @@ public function getCode()
         return $this->code;
     }
 
-    /**
-     * @return AkeneoLocale
-     */
     public function setCode(string $code): self
     {
         $this->code = $code;
@@ -93,9 +90,6 @@ public function getAkeneoSettings()
         return $this->akeneoSettings;
     }
 
-    /**
-     * @return AkeneoLocale
-     */
     public function setAkeneoSettings(AkeneoSettings $akeneoSettings = null): self
     {
         $this->akeneoSettings = $akeneoSettings;