Improvements

- show if batch message is not processed
- add env var to disable async mode
- improve reply message

Author: Maksym Kotliar

Async/ElasticaPopulateProcessor.php

@@ -32,32 +32,39 @@ public function process(Message $message, Context $context)
         }
 
         if ($message->isRedelivered()) {
-            $replyMessage = $context->createMessage(false);
-            $replyQueue = $context->createQueue($message->getReplyTo());
-            $context->createProducer()->send($replyQueue, $replyMessage);
+            $this->sendReply($context, $message, false);
 
             return self::REJECT;
         }
 
-        $options = JSON::decode($message->getBody());
+        try {
+            $options = JSON::decode($message->getBody());
 
-        $provider = $this->providerRegistry->getProvider($options['indexName'], $options['typeName']);
-        $provider->populate(null, $options);
+            $provider = $this->providerRegistry->getProvider($options['indexName'], $options['typeName']);
+            $provider->populate(null, $options);
 
-        $this->sendReply($context, $message->getReplyTo(), true);
+            $this->sendReply($context, $message, true);
 
-        return self::ACK;
+            return self::ACK;
+        } catch (\Exception $e) {
+            $this->sendReply($context, $message, false);
+
+            return self::REJECT;
+        }
     }
 
     /**
      * @param Context $context
-     * @param string $replyTo
-     * @param bool $message
+     * @param Message $message
+     * @param bool $successful
      */
-    private function sendReply(Context $context, $replyTo, $message)
+    private function sendReply(Context $context, Message $message, $successful)
     {
-        $replyMessage = $context->createMessage($message);
-        $replyQueue = $context->createQueue($replyTo);
+        $replyMessage = $context->createMessage($message->getBody(), $message->getProperties(), $message->getHeaders());
+        $replyMessage->setProperty('fos-populate-successful', (int) $successful);
+
+        $replyQueue = $context->createQueue($message->getReplyTo());
+
         $context->createProducer()->send($replyQueue, $replyMessage);
     }
 }

Elastica/AsyncDoctrineOrmProvider.php

@@ -27,6 +27,10 @@ public function setContext(Context $context)
      */
     protected function doPopulate($options, \Closure $loggerClosure = null)
     {
+        if (getenv('ENQUEUE_ELASTICA_DISABLE_ASYNC')) {
+            return parent::doPopulate($options, $loggerClosure);
+        }
+
         $this->batchSize = null;
         if ($options['real_populate']) {
             $this->batchSize = $options['offset'] + $options['batch_size'];
@@ -58,8 +62,18 @@ protected function doPopulate($options, \Closure $loggerClosure = null)
         $limitTime = time() + 180;
         while ($nbMessages) {
             if ($message = $consumer->receive(20000)) {
-                if (null !== $loggerClosure) {
-                    $loggerClosure($options['batch_size'], $nbObjects);
+                $errorMessage = null;
+
+                $errorMessage = null;
+                if (false == $message->getProperty('fos-populate-successful', false)) {
+                    $errorMessage = sprintf(
+                        '<error>Batch failed: </error> <comment>Failed to process message %s</comment>',
+                        $message->getBody()
+                    );
+                }
+
+                if ($loggerClosure) {
+                    $loggerClosure($options['batch_size'], $nbObjects, $errorMessage);
                 }
 
                 $consumer->acknowledge($message);

README.md

@@ -81,6 +81,12 @@ So you have to run as usual
 $ ./bin/console fos:elastica:populate 
 ```
 
+If you want to disable this behavior you can un register the bundle or use env var
+
+```bash
+$ ENQUEUE_ELASTICA_DISABLE_ASYNC=1 ./bin/console fos:elastica:populate 
+```
+
 and have pull of consumer commands run somewhere, run them as many as you'd like
 
 ```bash