Skip to content

Commit 54f69b7

Browse files
dxopsdxops
authored
AKM-34: Product filter breaks configurable products (oroinc#65)
AKM-31: Split products and product models filters - do not allow duplicate jobs
1 parent 5747619 commit 54f69b7

20 files changed

+591
-130
lines changed

‎Controller/ValidateConnectionController.php

+1-1
Original file line numberDiff line numberDiff line change
@@ -113,7 +113,7 @@ public function validateConnectionAction(Request $request, Channel $channel = nu
113113
'akeneoLocales' => $akeneoLocales,
114114
'success' => $success,
115115
'message' => $message,
116-
'currencyList' => $this->currencyProvider->getCurrencies(),
116+
'currencyList' => $this->currencyProvider->getCurrencyList(),
117117
]
118118
);
119119
}

‎Entity/AkeneoSettings.php

+26
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,12 @@ class AkeneoSettings extends Transport
8989
* @ORM\Column(name="akeneo_product_filter", type="text", nullable=true)
9090
*/
9191
protected $productFilter;
92+
/**
93+
* @var string
94+
*
95+
* @ORM\Column(name="akeneo_conf_product_filter", type="text", nullable=true)
96+
*/
97+
protected $configurableProductFilter;
9298
/**
9399
* @var string
94100
*
@@ -275,6 +281,26 @@ public function setProductFilter($productFilter)
275281
return $this;
276282
}
277283

284+
/**
285+
* @return string
286+
*/
287+
public function getConfigurableProductFilter()
288+
{
289+
return $this->configurableProductFilter;
290+
}
291+
292+
/**
293+
* @param string $configurableProductFilter
294+
*
295+
* @return self
296+
*/
297+
public function setConfigurableProductFilter($configurableProductFilter)
298+
{
299+
$this->configurableProductFilter = $configurableProductFilter;
300+
301+
return $this;
302+
}
303+
278304
/**
279305
* @return ParameterBag
280306
*/

‎Form/Extension/ChannelTypeExtension.php

+2
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,12 @@ class ChannelTypeExtension extends AbstractTypeExtension
1515
* @var array
1616
*/
1717
protected $connectorsOrder = [
18+
'brand',
1819
'category',
1920
'attribute',
2021
'attribute_family',
2122
'product',
23+
'configurable_product',
2224
];
2325

2426
/**

‎Form/Type/AkeneoSettingsType.php

+9
Original file line numberDiff line numberDiff line change
@@ -249,6 +249,15 @@ public function buildForm(FormBuilderInterface $builder, array $options)
249249
],
250250
]
251251
)
252+
->add(
253+
'configurableProductFilter',
254+
TextareaType::class,
255+
[
256+
'required' => false,
257+
'label' => 'oro.akeneo.integration.settings.akeneo_configurable_product_filter.label',
258+
'constraints' => [new JsonConstraint()],
259+
]
260+
)
252261
->add(
253262
'priceList',
254263
PriceListSelectType::class,

‎ImportExport/Processor/AsyncProcessor.php

-43
Original file line numberDiff line numberDiff line change
@@ -6,51 +6,8 @@
66

77
class AsyncProcessor implements ProcessorInterface
88
{
9-
use CacheProviderAwareProcessor;
10-
11-
/** @var array */
12-
private $variants = [];
13-
149
public function process($item)
1510
{
16-
$this->updateVariants($item);
17-
1811
return $item;
1912
}
20-
21-
private function updateVariants(array &$item)
22-
{
23-
$sku = $item['sku'];
24-
25-
if (!empty($item['family_variant'])) {
26-
if (isset($item['parent'], $this->variants[$sku])) {
27-
$parent = $item['parent'];
28-
foreach (array_keys($this->variants[$sku]) as $sku) {
29-
$this->variants[$parent][$sku] = ['parent' => $parent, 'variant' => $sku];
30-
}
31-
}
32-
33-
return;
34-
}
35-
36-
if (empty($item['parent'])) {
37-
return;
38-
}
39-
40-
$parent = $item['parent'];
41-
42-
$this->variants[$parent][$sku] = ['parent' => $parent, 'variant' => $sku];
43-
}
44-
45-
public function initialize()
46-
{
47-
$this->variants = [];
48-
$this->cacheProvider->delete('product_variants');
49-
}
50-
51-
public function flush()
52-
{
53-
$this->cacheProvider->save('product_variants', $this->variants);
54-
$this->variants = [];
55-
}
5613
}

‎ImportExport/Writer/AsyncWriter.php

+56-43
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55
use Doctrine\DBAL\Platforms\MySqlPlatform;
66
use Doctrine\DBAL\Types\Types;
77
use Oro\Bundle\AkeneoBundle\Async\Topics;
8-
use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
8+
use Oro\Bundle\AkeneoBundle\EventListener\AdditionalOptionalListenerManager;
99
use Oro\Bundle\BatchBundle\Entity\StepExecution;
1010
use Oro\Bundle\BatchBundle\Item\ItemWriterInterface;
1111
use Oro\Bundle\BatchBundle\Item\Support\ClosableInterface;
@@ -14,6 +14,7 @@
1414
use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges;
1515
use Oro\Bundle\MessageQueueBundle\Client\BufferedMessageProducer;
1616
use Oro\Bundle\MessageQueueBundle\Entity\Job;
17+
use Oro\Bundle\PlatformBundle\Manager\OptionalListenerManager;
1718
use Oro\Component\MessageQueue\Client\Message;
1819
use Oro\Component\MessageQueue\Client\MessagePriority;
1920
use Oro\Component\MessageQueue\Client\MessageProducerInterface;
@@ -23,10 +24,6 @@ class AsyncWriter implements
2324
ClosableInterface,
2425
StepExecutionAwareInterface
2526
{
26-
use CacheProviderTrait;
27-
28-
private const VARIANTS_BATCH_SIZE = 25;
29-
3027
/** @var MessageProducerInterface * */
3128
private $messageProducer;
3229

@@ -39,17 +36,30 @@ class AsyncWriter implements
3936
/** @var DoctrineHelper */
4037
private $doctrineHelper;
4138

39+
/** @var OptionalListenerManager */
40+
private $optionalListenerManager;
41+
42+
/** @var AdditionalOptionalListenerManager */
43+
private $additionalOptionalListenerManager;
44+
4245
public function __construct(
4346
MessageProducerInterface $messageProducer,
44-
DoctrineHelper $doctrineHelper
47+
DoctrineHelper $doctrineHelper,
48+
OptionalListenerManager $optionalListenerManager,
49+
AdditionalOptionalListenerManager $additionalOptionalListenerManager
4550
) {
4651
$this->messageProducer = $messageProducer;
4752
$this->doctrineHelper = $doctrineHelper;
53+
$this->optionalListenerManager = $optionalListenerManager;
54+
$this->additionalOptionalListenerManager = $additionalOptionalListenerManager;
4855
}
4956

5057
public function initialize()
5158
{
5259
$this->size = 0;
60+
61+
$this->additionalOptionalListenerManager->disableListeners();
62+
$this->optionalListenerManager->disableListeners($this->optionalListenerManager->getListeners());
5363
}
5464

5565
public function write(array $items)
@@ -67,52 +77,30 @@ public function write(array $items)
6777
$this->stepExecution->setWriteCount($this->size);
6878

6979
$jobId = $this->insertJob($jobName);
70-
$this->createFieldsChanges($jobId, $items, 'items');
71-
$this->sendMessage($channelId, $jobId, true);
72-
}
73-
74-
public function flush()
75-
{
76-
$this->size = 0;
77-
78-
$variants = $this->cacheProvider->fetch('product_variants') ?? [];
79-
if (!$variants) {
80-
return;
81-
}
82-
83-
$channelId = $this->stepExecution->getJobExecution()->getExecutionContext()->get('channel');
84-
85-
$chunks = array_chunk($variants, self::VARIANTS_BATCH_SIZE, true);
86-
87-
foreach ($chunks as $key => $chunk) {
88-
$jobName = sprintf(
89-
'oro_integration:sync_integration:%s:variants:%s-%s',
90-
$channelId,
91-
self::VARIANTS_BATCH_SIZE * $key + 1,
92-
self::VARIANTS_BATCH_SIZE * $key + count($chunk)
93-
);
94-
95-
$jobId = $this->insertJob($jobName);
96-
$this->createFieldsChanges($jobId, $chunk, 'variants');
97-
$this->sendMessage($channelId, $jobId);
80+
if ($jobId && $this->createFieldsChanges($jobId, $items, 'items')) {
81+
$this->sendMessage($channelId, $jobId, true);
9882
}
9983
}
10084

101-
private function createFieldsChanges(int $jobId, array &$data, string $key): void
85+
private function createFieldsChanges(int $jobId, array &$data, string $key): bool
10286
{
10387
$em = $this->doctrineHelper->getEntityManager(FieldsChanges::class);
10488
$fieldsChanges = $em
10589
->getRepository(FieldsChanges::class)
10690
->findOneBy(['entityId' => $jobId, 'entityClass' => Job::class]);
107-
if (!$fieldsChanges) {
108-
$fieldsChanges = new FieldsChanges([]);
109-
$fieldsChanges->setEntityClass(Job::class);
110-
$fieldsChanges->setEntityId($jobId);
91+
if ($fieldsChanges) {
92+
return false;
11193
}
94+
95+
$fieldsChanges = new FieldsChanges([]);
96+
$fieldsChanges->setEntityClass(Job::class);
97+
$fieldsChanges->setEntityId($jobId);
11298
$fieldsChanges->setChangedFields([$key => $data]);
11399
$em->persist($fieldsChanges);
114100
$em->flush($fieldsChanges);
115101
$em->clear(FieldsChanges::class);
102+
103+
return true;
116104
}
117105

118106
private function sendMessage(int $channelId, int $jobId, bool $incrementedRead = false): void
@@ -143,12 +131,15 @@ private function getRootJob(): ?int
143131
throw new \InvalidArgumentException('Root job id is empty');
144132
}
145133

146-
return $rootJobId;
134+
return (int)$rootJobId;
147135
}
148136

149137
public function close()
150138
{
151139
$this->size = 0;
140+
141+
$this->optionalListenerManager->enableListeners($this->optionalListenerManager->getListeners());
142+
$this->additionalOptionalListenerManager->enableListeners();
152143
}
153144

154145
public function setStepExecution(StepExecution $stepExecution)
@@ -159,12 +150,34 @@ public function setStepExecution(StepExecution $stepExecution)
159150
private function insertJob(string $jobName): ?int
160151
{
161152
$em = $this->doctrineHelper->getEntityManager(Job::class);
162-
$tableName = $em->getClassMetadata(Job::class)->getTableName();
163153
$connection = $em->getConnection();
154+
$rootJobId = $this->getRootJob();
155+
156+
$hasRootJob = $connection
157+
->executeStatement(
158+
'SELECT 1 FROM oro_message_queue_job WHERE id = :id LIMIT 1;',
159+
['id' => $rootJobId],
160+
['id' => Types::INTEGER]
161+
);
162+
163+
if (!$hasRootJob) {
164+
throw new \InvalidArgumentException(sprintf('Root job "%d" missing', $rootJobId));
165+
}
166+
167+
$childJob = $connection
168+
->executeStatement(
169+
'SELECT id FROM oro_message_queue_job WHERE root_job_id = :rootJob and name = :name LIMIT 1;',
170+
['rootJob' => $rootJobId, 'name' => $jobName],
171+
['rootJob' => Types::INTEGER, 'name' => Types::STRING]
172+
);
173+
174+
if ($childJob) {
175+
return $childJob;
176+
}
164177

165178
$qb = $connection->createQueryBuilder();
166179
$qb
167-
->insert($tableName)
180+
->insert('oro_message_queue_job')
168181
->values([
169182
'name' => ':name',
170183
'status' => ':status',
@@ -178,7 +191,7 @@ private function insertJob(string $jobName): ?int
178191
'interrupted' => false,
179192
'unique' => false,
180193
'createdAt' => new \DateTime(),
181-
'rootJob' => $this->getRootJob(),
194+
'rootJob' => $rootJobId,
182195
], [
183196
'name' => Types::STRING,
184197
'status' => Types::STRING,

0 commit comments

Comments
 (0)