|
| 1 | +<?php |
| 2 | + |
| 3 | +namespace Oro\Bundle\AkeneoBundle\ImportExport\Writer; |
| 4 | + |
| 5 | +use Doctrine\DBAL\Platforms\MySqlPlatform; |
| 6 | +use Doctrine\DBAL\Types\Types; |
| 7 | +use Oro\Bundle\AkeneoBundle\Async\Topics; |
| 8 | +use Oro\Bundle\AkeneoBundle\EventListener\AdditionalOptionalListenerManager; |
| 9 | +use Oro\Bundle\BatchBundle\Entity\StepExecution; |
| 10 | +use Oro\Bundle\BatchBundle\Item\ItemWriterInterface; |
| 11 | +use Oro\Bundle\BatchBundle\Step\StepExecutionAwareInterface; |
| 12 | +use Oro\Bundle\EntityBundle\ORM\DoctrineHelper; |
| 13 | +use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges; |
| 14 | +use Oro\Bundle\MessageQueueBundle\Client\BufferedMessageProducer; |
| 15 | +use Oro\Bundle\MessageQueueBundle\Entity\Job; |
| 16 | +use Oro\Bundle\PlatformBundle\Manager\OptionalListenerManager; |
| 17 | +use Oro\Component\MessageQueue\Client\Message; |
| 18 | +use Oro\Component\MessageQueue\Client\MessagePriority; |
| 19 | +use Oro\Component\MessageQueue\Client\MessageProducerInterface; |
| 20 | + |
| 21 | +class ConfigurableAsyncWriter implements |
| 22 | + ItemWriterInterface, |
| 23 | + StepExecutionAwareInterface |
| 24 | +{ |
| 25 | + private const VARIANTS_BATCH_SIZE = 25; |
| 26 | + |
| 27 | + /** @var MessageProducerInterface * */ |
| 28 | + private $messageProducer; |
| 29 | + |
| 30 | + /** @var StepExecution */ |
| 31 | + private $stepExecution; |
| 32 | + |
| 33 | + /** @var DoctrineHelper */ |
| 34 | + private $doctrineHelper; |
| 35 | + |
| 36 | + /** @var OptionalListenerManager */ |
| 37 | + private $optionalListenerManager; |
| 38 | + |
| 39 | + /** @var AdditionalOptionalListenerManager */ |
| 40 | + private $additionalOptionalListenerManager; |
| 41 | + |
| 42 | + private $variants = []; |
| 43 | + |
| 44 | + public function __construct( |
| 45 | + MessageProducerInterface $messageProducer, |
| 46 | + DoctrineHelper $doctrineHelper, |
| 47 | + OptionalListenerManager $optionalListenerManager, |
| 48 | + AdditionalOptionalListenerManager $additionalOptionalListenerManager |
| 49 | + ) { |
| 50 | + $this->messageProducer = $messageProducer; |
| 51 | + $this->doctrineHelper = $doctrineHelper; |
| 52 | + $this->optionalListenerManager = $optionalListenerManager; |
| 53 | + $this->additionalOptionalListenerManager = $additionalOptionalListenerManager; |
| 54 | + } |
| 55 | + |
| 56 | + public function initialize() |
| 57 | + { |
| 58 | + $this->variants = []; |
| 59 | + |
| 60 | + $this->additionalOptionalListenerManager->disableListeners(); |
| 61 | + $this->optionalListenerManager->disableListeners($this->optionalListenerManager->getListeners()); |
| 62 | + } |
| 63 | + |
| 64 | + public function write(array $items) |
| 65 | + { |
| 66 | + foreach ($items as $item) { |
| 67 | + $sku = $item['sku']; |
| 68 | + |
| 69 | + if (!empty($item['family_variant'])) { |
| 70 | + if (isset($item['parent'], $this->variants[$sku])) { |
| 71 | + $parent = $item['parent']; |
| 72 | + foreach (array_keys($this->variants[$sku]) as $sku) { |
| 73 | + $this->variants[$parent][$sku] = ['parent' => $parent, 'variant' => $sku]; |
| 74 | + } |
| 75 | + } |
| 76 | + |
| 77 | + return; |
| 78 | + } |
| 79 | + |
| 80 | + if (empty($item['parent'])) { |
| 81 | + return; |
| 82 | + } |
| 83 | + |
| 84 | + $parent = $item['parent']; |
| 85 | + |
| 86 | + $this->variants[$parent][$sku] = ['parent' => $parent, 'variant' => $sku]; |
| 87 | + } |
| 88 | + } |
| 89 | + |
| 90 | + public function close() |
| 91 | + { |
| 92 | + $this->variants = []; |
| 93 | + |
| 94 | + $this->optionalListenerManager->enableListeners($this->optionalListenerManager->getListeners()); |
| 95 | + $this->additionalOptionalListenerManager->enableListeners(); |
| 96 | + } |
| 97 | + |
| 98 | + public function flush() |
| 99 | + { |
| 100 | + $channelId = $this->stepExecution->getJobExecution()->getExecutionContext()->get('channel'); |
| 101 | + |
| 102 | + $chunks = array_chunk($this->variants, self::VARIANTS_BATCH_SIZE, true); |
| 103 | + |
| 104 | + foreach ($chunks as $key => $chunk) { |
| 105 | + $jobName = sprintf( |
| 106 | + 'oro_integration:sync_integration:%s:variants:%s-%s', |
| 107 | + $channelId, |
| 108 | + self::VARIANTS_BATCH_SIZE * $key + 1, |
| 109 | + self::VARIANTS_BATCH_SIZE * $key + count($chunk) |
| 110 | + ); |
| 111 | + |
| 112 | + $jobId = $this->insertJob($jobName); |
| 113 | + if ($jobId && $this->createFieldsChanges($jobId, $chunk, 'variants')) { |
| 114 | + $this->sendMessage($channelId, $jobId); |
| 115 | + } |
| 116 | + } |
| 117 | + } |
| 118 | + |
| 119 | + private function createFieldsChanges(int $jobId, array &$data, string $key): bool |
| 120 | + { |
| 121 | + $em = $this->doctrineHelper->getEntityManager(FieldsChanges::class); |
| 122 | + $fieldsChanges = $em |
| 123 | + ->getRepository(FieldsChanges::class) |
| 124 | + ->findOneBy(['entityId' => $jobId, 'entityClass' => Job::class]); |
| 125 | + if ($fieldsChanges) { |
| 126 | + return false; |
| 127 | + } |
| 128 | + |
| 129 | + $fieldsChanges = new FieldsChanges([]); |
| 130 | + $fieldsChanges->setEntityClass(Job::class); |
| 131 | + $fieldsChanges->setEntityId($jobId); |
| 132 | + $fieldsChanges->setChangedFields([$key => $data]); |
| 133 | + $em->persist($fieldsChanges); |
| 134 | + $em->flush($fieldsChanges); |
| 135 | + $em->clear(FieldsChanges::class); |
| 136 | + |
| 137 | + return true; |
| 138 | + } |
| 139 | + |
| 140 | + private function sendMessage(int $channelId, int $jobId, bool $incrementedRead = false): void |
| 141 | + { |
| 142 | + $this->messageProducer->send( |
| 143 | + Topics::IMPORT_PRODUCTS, |
| 144 | + new Message( |
| 145 | + [ |
| 146 | + 'integrationId' => $channelId, |
| 147 | + 'jobId' => $jobId, |
| 148 | + 'connector' => 'configurable_product', |
| 149 | + 'connector_parameters' => ['incremented_read' => $incrementedRead], |
| 150 | + ], |
| 151 | + MessagePriority::HIGH |
| 152 | + ) |
| 153 | + ); |
| 154 | + |
| 155 | + if ($this->messageProducer instanceof BufferedMessageProducer |
| 156 | + && $this->messageProducer->isBufferingEnabled()) { |
| 157 | + $this->messageProducer->flushBuffer(); |
| 158 | + } |
| 159 | + } |
| 160 | + |
| 161 | + private function getRootJob(): ?int |
| 162 | + { |
| 163 | + $rootJobId = $this->stepExecution->getJobExecution()->getExecutionContext()->get('rootJobId') ?? null; |
| 164 | + if (!$rootJobId) { |
| 165 | + throw new \InvalidArgumentException('Root job id is empty'); |
| 166 | + } |
| 167 | + |
| 168 | + return (int)$rootJobId; |
| 169 | + } |
| 170 | + |
| 171 | + public function setStepExecution(StepExecution $stepExecution) |
| 172 | + { |
| 173 | + $this->stepExecution = $stepExecution; |
| 174 | + } |
| 175 | + |
| 176 | + private function insertJob(string $jobName): ?int |
| 177 | + { |
| 178 | + $em = $this->doctrineHelper->getEntityManager(Job::class); |
| 179 | + $connection = $em->getConnection(); |
| 180 | + $rootJobId = $this->getRootJob(); |
| 181 | + |
| 182 | + $hasRootJob = $connection |
| 183 | + ->executeStatement( |
| 184 | + 'SELECT 1 FROM oro_message_queue_job WHERE id = :id LIMIT 1;', |
| 185 | + ['id' => $rootJobId], |
| 186 | + ['id' => Types::INTEGER] |
| 187 | + ); |
| 188 | + |
| 189 | + if (!$hasRootJob) { |
| 190 | + throw new \InvalidArgumentException(sprintf('Root job "%d" missing', $rootJobId)); |
| 191 | + } |
| 192 | + |
| 193 | + $childJob = $connection |
| 194 | + ->executeStatement( |
| 195 | + 'SELECT id FROM oro_message_queue_job WHERE root_job_id = :rootJob and name = :name LIMIT 1;', |
| 196 | + ['rootJob' => $rootJobId, 'name' => $jobName], |
| 197 | + ['rootJob' => Types::INTEGER, 'name' => Types::STRING] |
| 198 | + ); |
| 199 | + |
| 200 | + if ($childJob) { |
| 201 | + return $childJob; |
| 202 | + } |
| 203 | + |
| 204 | + $qb = $connection->createQueryBuilder(); |
| 205 | + $qb |
| 206 | + ->insert('oro_message_queue_job') |
| 207 | + ->values([ |
| 208 | + 'name' => ':name', |
| 209 | + 'status' => ':status', |
| 210 | + 'interrupted' => ':interrupted', |
| 211 | + 'created_at' => ':createdAt', |
| 212 | + 'root_job_id' => ':rootJob', |
| 213 | + ]) |
| 214 | + ->setParameters([ |
| 215 | + 'name' => $jobName, |
| 216 | + 'status' => Job::STATUS_NEW, |
| 217 | + 'interrupted' => false, |
| 218 | + 'unique' => false, |
| 219 | + 'createdAt' => new \DateTime(), |
| 220 | + 'rootJob' => $rootJobId, |
| 221 | + ], [ |
| 222 | + 'name' => Types::STRING, |
| 223 | + 'status' => Types::STRING, |
| 224 | + 'interrupted' => Types::BOOLEAN, |
| 225 | + 'unique' => Types::BOOLEAN, |
| 226 | + 'createdAt' => Types::DATETIME_MUTABLE, |
| 227 | + 'rootJob' => Types::INTEGER, |
| 228 | + ]); |
| 229 | + |
| 230 | + if ($connection->getDatabasePlatform() instanceof MySqlPlatform) { |
| 231 | + $qb->setValue('`unique`', ':unique'); |
| 232 | + } else { |
| 233 | + $qb->setValue('"unique"', ':unique'); |
| 234 | + } |
| 235 | + |
| 236 | + $qb->execute(); |
| 237 | + |
| 238 | + return $connection->lastInsertId(); |
| 239 | + } |
| 240 | +} |
0 commit comments