fix catching exceptions

Author: anvasiliev

app/code/Magento/AmqpStore/Plugin/AsynchronousOperations/MassConsumerEnvelopeCallback.php

@@ -51,12 +51,17 @@ public function __construct(
     }
 
     /**
+     * Check if amqpProperties['application_headers'] have 'store_id' and use it to setCurrentStore
+     * Restore currentStore of consumer process after execution.
+     *
      * @param SubjectMassConsumerEnvelopeCallback $subject
+     * @param callable $proceed
      * @param EnvelopeInterface $message
      * @return array|null
+     * @throws NoSuchEntityException
      * @SuppressWarnings(PHPMD.UnusedFormalParameter)
      */
-    public function beforeExecute(SubjectMassConsumerEnvelopeCallback $subject, EnvelopeInterface $message)
+    public function aroundExecute(SubjectMassConsumerEnvelopeCallback $subject, callable $proceed, EnvelopeInterface $message)
     {
         $amqpProperties = $message->getProperties();
         if (isset($amqpProperties['application_headers'])) {
@@ -72,13 +77,17 @@ public function beforeExecute(SubjectMassConsumerEnvelopeCallback $subject, Enve
                     $this->logger->error(
                         sprintf("Can't set currentStoreId during processing queue. Error %s.", $e->getMessage())
                     );
-                    return null;
+                    throw new NoSuchEntityException(__($e->getMessage()));
                 }
                 if (isset($storeId) && $storeId !== $currentStoreId) {
                     $this->storeManager->setCurrentStore($storeId);
                 }
             }
         }
-        return [$message];
+        $result = $proceed($message);
+        if (isset($storeId, $currentStoreId) && $storeId !== $currentStoreId) {
+            $this->storeManager->setCurrentStore($currentStoreId);//restore previous current store
+        }
+        return $result;
     }
 }

app/code/Magento/AmqpStore/Plugin/Framework/Amqp/Bulk/Exchange.php

@@ -55,17 +55,22 @@ public function __construct(
      * @param $topic
      * @param EnvelopeInterface[] $envelopes
      * @return array|null
+     * @throws NoSuchEntityException
+     * @throws AMQPInvalidArgumentException
      * @SuppressWarnings(PHPMD.UnusedFormalParameter)
      */
     public function beforeEnqueue(SubjectExchange $subject, $topic, array $envelopes)
     {
         try {
             $storeId = $this->storeManager->getStore()->getId();
+            $code = $this->storeManager->getStore()->getCode();
+            $t=1;
         } catch (NoSuchEntityException $e) {
+            $r=1;
             $this->logger->error(
                 sprintf("Can't get current storeId and inject to amqp message. Error %s.", $e->getMessage())
             );
-            return null;
+            throw new NoSuchEntityException(__($e->getMessage()));
         }
 
         $updatedEnvelopes = [];
@@ -83,7 +88,7 @@ public function beforeEnqueue(SubjectExchange $subject, $topic, array $envelopes
                         $this->logger->error(
                             sprintf("Can't set storeId to amqp message. Error %s.", $ea->getMessage())
                         );
-                        return null;
+                        throw new AMQPInvalidArgumentException($ea->getMessage());
                     }
                     $properties['application_headers'] = $headers;
                 }