Merge pull request #21 from aumel/issue/902

Invoke sendUpdateIndexMessage on PostFlush

Author: makasim

Doctrine/SyncIndexWithObjectChangeListener.php

@@ -2,6 +2,7 @@
 namespace Enqueue\ElasticaBundle\Doctrine;
 
 use Doctrine\Common\Persistence\Event\LifecycleEventArgs;
+use Doctrine\ORM\Event\PostFlushEventArgs;
 use Enqueue\ElasticaBundle\Doctrine\Queue\Commands;
 use Enqueue\ElasticaBundle\Doctrine\Queue\SyncIndexWithObjectChangeProcessor as SyncProcessor;
 use Enqueue\Util\JSON;
@@ -17,6 +18,11 @@ final class SyncIndexWithObjectChangeListener implements EventSubscriber
      */
     private $modelClass;
 
+    /**
+     * @var array
+     */
+    private $scheduledForUpdateIndex = [];
+
     /**
      * @var array
      */
@@ -31,23 +37,33 @@ public function __construct(Context $context, $modelClass, array $config)
 
     public function postUpdate(LifecycleEventArgs $args)
     {
-
         if ($args->getObject() instanceof $this->modelClass) {
-            $this->sendUpdateIndexMessage(SyncProcessor::UPDATE_ACTION, $args);
+            $this->scheduledForUpdateIndex[] = ['action' => SyncProcessor::UPDATE_ACTION, 'args' => $args];
         }
     }
 
     public function postPersist(LifecycleEventArgs $args)
     {
         if ($args->getObject() instanceof $this->modelClass) {
-            $this->sendUpdateIndexMessage(SyncProcessor::INSERT_ACTION, $args);
+            $this->scheduledForUpdateIndex[] = ['action' => SyncProcessor::INSERT_ACTION, 'args' => $args];
         }
     }
 
     public function preRemove(LifecycleEventArgs $args)
     {
         if ($args->getObject() instanceof $this->modelClass) {
-            $this->sendUpdateIndexMessage(SyncProcessor::REMOVE_ACTION, $args);
+            $this->scheduledForUpdateIndex[] = ['action' => SyncProcessor::REMOVE_ACTION, 'args' => $args];
+        }
+    }
+
+    public function postFlush(PostFlushEventArgs $event)
+    {
+        if (count($this->scheduledForUpdateIndex)) {
+            foreach ($this->scheduledForUpdateIndex as $updateIndex) {
+                $this->sendUpdateIndexMessage($updateIndex['action'], $updateIndex['args']);
+            }
+
+            $this->scheduledForUpdateIndex = [];
         }
     }
 
@@ -57,6 +73,7 @@ public function getSubscribedEvents()
             'postPersist',
             'postUpdate',
             'preRemove',
+            'postFlush'
         ];
     }