storeManager = $storeManager; $this->messageEncoder = $messageEncoder; $this->messageValidator = $messageValidator; $this->jsonHelper = $jsonHelper; $this->logger = $logger; } /** * Check if message contains 'store_id' and use it to setCurrentStore * Restore original store value in consumer process after execution. * Reject queue messages because of wrong store_id. * * @param SubjectMassConsumerEnvelopeCallback $subject * @param callable $proceed * @param EnvelopeInterface $message * @return void * @SuppressWarnings(PHPMD.UnusedFormalParameter) */ public function aroundExecute( SubjectMassConsumerEnvelopeCallback $subject, callable $proceed, EnvelopeInterface $message ) { /** @var OperationInterface $operation */ $operation = $this->messageEncoder->decode(AsyncConfig::SYSTEM_TOPIC_NAME, $message->getBody()); $this->messageValidator->validate(AsyncConfig::SYSTEM_TOPIC_NAME, $operation); $data = $this->jsonHelper->unserialize($operation->getSerializedData()); if (isset($data['store_id'])) { $storeId = $data['store_id']; try { $currentStoreId = $this->storeManager->getStore()->getId(); } catch (NoSuchEntityException $e) { $this->logger->error( sprintf( "Can't set currentStoreId during processing queue. Message rejected. Error %s.", $e->getMessage() ) ); $subject->getQueue()->reject($message, false, $e->getMessage()); return; } if ($storeId !== $currentStoreId) { $this->storeManager->setCurrentStore($storeId); } } $proceed($message); if (isset($storeId, $currentStoreId) && $storeId !== $currentStoreId) { $this->storeManager->setCurrentStore($currentStoreId);//restore original store value } } }