track entities (#5)

* Track persisted entities and update their ids when they are created in ES

* Fixed Symfony version to 3.2.* as 3.3.0 is breaking the KNP paginator dependency on the translator service

* Raised phpcs version to 3.*

* code style fixes

* Updated .travis.yml

.travis.yml

@@ -6,7 +6,7 @@ php:
   - hhvm
 
 env:
-  - SYMFONY_VERSION="~3.0" ES_VERSION="2.3.2"
+  - SYMFONY_VERSION="3.2.*" ES_VERSION="2.3.2"
 
 matrix:
   allow_failures:
@@ -15,13 +15,13 @@ matrix:
     - php: 5.6
       env: SYMFONY_VERSION="~2.7" ES_VERSION="1.6.2"
     - php: 7.0
-      env: SYMFONY_VERSION="~3.0" ES_VERSION="1.7.4"
+      env: SYMFONY_VERSION="3.2.*" ES_VERSION="1.7.4"
     - php: 7.0
-      env: SYMFONY_VERSION="~3.0" ES_VERSION="1.6.2"
+      env: SYMFONY_VERSION="3.2.*" ES_VERSION="1.6.2"
     - php: 7.0
-      env: SYMFONY_VERSION="~3.0" ES_VERSION="2.1.1"
+      env: SYMFONY_VERSION="3.2.*" ES_VERSION="2.1.1"
     - php: 7.0
-      env: SYMFONY_VERSION="~3.0" ES_VERSION="2.2.2"
+      env: SYMFONY_VERSION="3.2.*" ES_VERSION="2.2.2"
 
 install:
   - composer require --no-update symfony/symfony:${SYMFONY_VERSION}

DependencyInjection/Compiler/AddIndexManagersPass.php

@@ -28,7 +28,7 @@ public function process(ContainerBuilder $container)
             $connectionService = sprintf('sfes.connection.%s', $indexSettings['connection']);
             if (!$container->hasDefinition($connectionService)) {
                 throw new InvalidConfigurationException(
-                    'There is no ES connection with name ' . $indexSettings['connection']
+                    'There is no ES connection with name '.$indexSettings['connection']
                 );
             }
 
@@ -39,7 +39,7 @@ public function process(ContainerBuilder $container)
                     $indexManagerClass,
                     $indexManagerName,
                     $container->getDefinition($connectionService),
-                    $indexSettings
+                    $indexSettings,
                 ]
             );
 

Event/Events.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace Sineflow\ElasticsearchBundle\Event;
+
+/**
+ * Class Events
+ */
+final class Events
+{
+    /**
+     * Dispatched on persisting an entity via the index manager
+     */
+    const PRE_PERSIST = 'sfes.pre_persist';
+
+    /**
+     * Dispatched after a bulk request is submitted to Elasticsearch
+     */
+    const POST_COMMIT = 'sfes.post_commit';
+}

Event/PostCommitEvent.php

@@ -0,0 +1,32 @@
+<?php
+
+namespace Sineflow\ElasticsearchBundle\Event;
+
+use Symfony\Component\EventDispatcher\Event;
+
+/**
+ * Class PostCommitEvent
+ */
+class PostCommitEvent extends Event
+{
+    /**
+     * @var array
+     */
+    private $bulkResponse;
+
+    /**
+     * @param array $bulkResponse
+     */
+    public function __construct(array $bulkResponse)
+    {
+        $this->bulkResponse = $bulkResponse;
+    }
+
+    /**
+     * @return array
+     */
+    public function getBulkResponse()
+    {
+        return $this->bulkResponse;
+    }
+}

Event/PrePersistEvent.php

@@ -0,0 +1,48 @@
+<?php
+
+namespace Sineflow\ElasticsearchBundle\Event;
+
+use Sineflow\ElasticsearchBundle\Document\DocumentInterface;
+use Symfony\Component\EventDispatcher\Event;
+
+/**
+ * Class PrePersistEvent
+ */
+class PrePersistEvent extends Event
+{
+    /**
+     * @var DocumentInterface
+     */
+    private $document;
+
+    /**
+     * @var int
+     */
+    private $bulkOperationIndex;
+
+    /**
+     * @param DocumentInterface $document
+     * @param int               $bulkOperationIndex
+     */
+    public function __construct(DocumentInterface $document, $bulkOperationIndex)
+    {
+        $this->document           = $document;
+        $this->bulkOperationIndex = $bulkOperationIndex;
+    }
+
+    /**
+     * @return DocumentInterface
+     */
+    public function getDocument()
+    {
+        return $this->document;
+    }
+
+    /**
+     * @return int
+     */
+    public function getBulkOperationIndex()
+    {
+        return $this->bulkOperationIndex;
+    }
+}

Manager/ConnectionManager.php

@@ -6,7 +6,10 @@
 use Elasticsearch\Common\Exceptions\InvalidArgumentException;
 use Psr\Log\LoggerInterface;
 use Sineflow\ElasticsearchBundle\DTO\BulkQueryItem;
+use Sineflow\ElasticsearchBundle\Event\Events;
+use Sineflow\ElasticsearchBundle\Event\PostCommitEvent;
 use Sineflow\ElasticsearchBundle\Exception\BulkRequestException;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 /**
  * This class interacts with elasticsearch using injected client.
@@ -48,6 +51,11 @@ class ConnectionManager
      */
     private $autocommit;
 
+    /**
+     * @var EventDispatcherInterface
+     */
+    private $eventDispatcher;
+
     /**
      * Construct.
      *
@@ -126,6 +134,22 @@ public function setAutocommit($autocommit)
         $this->autocommit = $autocommit;
     }
 
+    /**
+     * @return EventDispatcherInterface
+     */
+    public function getEventDispatcher()
+    {
+        return $this->eventDispatcher;
+    }
+
+    /**
+     * @param EventDispatcherInterface $eventDispatcher
+     */
+    public function setEventDispatcher($eventDispatcher)
+    {
+        $this->eventDispatcher = $eventDispatcher;
+    }
+
     /**
      * Adds query to bulk queries container.
      *
@@ -141,6 +165,16 @@ public function addBulkOperation($operation, $index, $type, array $query)
         $this->bulkQueries[] = new BulkQueryItem($operation, $index, $type, $query);
     }
 
+    /**
+     * Returns the number of bulk operations currently queued
+     *
+     * @return int
+     */
+    public function getBulkOperationsCount()
+    {
+        return count($this->bulkQueries);
+    }
+
     /**
      * Optional setter to change bulk query params.
      *
@@ -203,6 +237,10 @@ public function commit($forceRefresh = true)
             $e->setBulkResponseItems($response['items']);
             throw $e;
         }
+
+        if ($this->eventDispatcher) {
+            $this->eventDispatcher->dispatch(Events::POST_COMMIT, new PostCommitEvent($response));
+        }
     }
 
     /**

Manager/ConnectionManagerFactory.php

@@ -5,13 +5,13 @@
 use Elasticsearch\ClientBuilder;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 /**
  * Elasticsearch connection factory class
  */
 class ConnectionManagerFactory
 {
-
     /**
      * @var LoggerInterface
      */
@@ -27,6 +27,11 @@ class ConnectionManagerFactory
      */
     private $kernelDebug;
 
+    /**
+     * @var EventDispatcherInterface
+     */
+    private $eventDispatcher;
+
     /**
      * @param boolean         $kernelDebug
      * @param LoggerInterface $tracer
@@ -39,6 +44,22 @@ public function __construct($kernelDebug, LoggerInterface $tracer = null, Logger
         $this->logger = $logger;
     }
 
+    /**
+     * @return EventDispatcherInterface
+     */
+    public function getEventDispatcher()
+    {
+        return $this->eventDispatcher;
+    }
+
+    /**
+     * @param EventDispatcherInterface $eventDispatcher
+     */
+    public function setEventDispatcher($eventDispatcher)
+    {
+        $this->eventDispatcher = $eventDispatcher;
+    }
+
     /**
      * @param string $connectionName
      * @param array  $connectionSettings
@@ -65,6 +86,7 @@ public function createConnectionManager($connectionName, $connectionSettings)
         );
 
         $connectionManager->setLogger($this->logger ?: new NullLogger());
+        $connectionManager->setEventDispatcher($this->eventDispatcher);
 
         return $connectionManager;
     }

Manager/IndexManager.php

@@ -7,13 +7,16 @@
 use Sineflow\ElasticsearchBundle\Document\Provider\ProviderInterface;
 use Sineflow\ElasticsearchBundle\Document\Provider\ProviderRegistry;
 use Sineflow\ElasticsearchBundle\Document\Repository\Repository;
+use Sineflow\ElasticsearchBundle\Event\Events;
+use Sineflow\ElasticsearchBundle\Event\PrePersistEvent;
 use Sineflow\ElasticsearchBundle\Exception\BulkRequestException;
 use Sineflow\ElasticsearchBundle\Exception\Exception;
 use Sineflow\ElasticsearchBundle\Exception\IndexRebuildingException;
 use Sineflow\ElasticsearchBundle\Exception\NoReadAliasException;
 use Sineflow\ElasticsearchBundle\Finder\Finder;
 use Sineflow\ElasticsearchBundle\Mapping\DocumentMetadataCollector;
 use Sineflow\ElasticsearchBundle\Result\DocumentConverter;
+use Symfony\Component\EventDispatcher\EventDispatcherInterface;
 
 /**
  * Manager class.
@@ -56,7 +59,7 @@ class IndexManager
     protected $indexMapping;
 
     /**
-     * @var RepositoryInterface[]
+     * @var Repository[]
      */
     protected $repositories = [];
 
@@ -75,6 +78,11 @@ class IndexManager
      */
     protected $writeAlias = null;
 
+    /**
+     * @var EventDispatcherInterface
+     */
+    protected $eventDispatcher;
+
     /**
      * @param string                    $managerName
      * @param ConnectionManager         $connection
@@ -110,6 +118,22 @@ public function __construct(
         }
     }
 
+    /**
+     * @return EventDispatcherInterface
+     */
+    public function getEventDispatcher()
+    {
+        return $this->eventDispatcher;
+    }
+
+    /**
+     * @param EventDispatcherInterface $eventDispatcher
+     */
+    public function setEventDispatcher($eventDispatcher)
+    {
+        $this->eventDispatcher = $eventDispatcher;
+    }
+
     /**
      * Returns mapping array for index
      *
@@ -248,12 +272,12 @@ private function getBaseIndexName()
      */
     private function getUniqueIndexName()
     {
-        $indexName = $baseName = $this->getBaseIndexName() . '_' . date('YmdHis');
+        $indexName = $baseName = $this->getBaseIndexName().'_'.date('YmdHis');
 
         $i = 1;
         // Keep trying other names until there is no such existing index or alias
         while ($this->getConnection()->existsIndexOrAlias(array('index' => $indexName))) {
-            $indexName = $baseName . '_' . $i;
+            $indexName = $baseName.'_'.$i;
             $i++;
         }
 
@@ -474,9 +498,9 @@ public function rebuildIndex($deleteOld = false, $cancelExistingRebuild = false)
                         [
                             'remove' => [
                                 'index' => $oldIndex,
-                                'alias' => $this->writeAlias
+                                'alias' => $this->writeAlias,
                             ],
-                        ]
+                        ],
                     ],
                 ],
             ];
@@ -655,7 +679,6 @@ public function update($documentClass, $id, array $fields = [], $script = null,
         }
     }
 
-
     /**
      * Adds document to a bulk request for the next commit.
      * Depending on the connection autocommit mode, the update may be committed right away.
@@ -664,6 +687,11 @@ public function update($documentClass, $id, array $fields = [], $script = null,
      */
     public function persist(DocumentInterface $document)
     {
+        if ($this->eventDispatcher) {
+            $bulkOperationIndex = $this->getConnection()->getBulkOperationsCount();
+            $this->eventDispatcher->dispatch(Events::PRE_PERSIST, new PrePersistEvent($document, $bulkOperationIndex));
+        }
+
         $documentArray = $this->documentConverter->convertToArray($document);
         $this->persistRaw(get_class($document), $documentArray);
     }