305 lines
9.3 KiB
PHP
305 lines
9.3 KiB
PHP
<?php
|
|
|
|
namespace FOS\ElasticaBundle\Doctrine;
|
|
|
|
use Psr\Log\LoggerInterface;
|
|
use Doctrine\Common\EventArgs;
|
|
use Doctrine\Common\EventSubscriber;
|
|
use FOS\ElasticaBundle\Persister\ObjectPersisterInterface;
|
|
use FOS\ElasticaBundle\Persister\ObjectPersister;
|
|
use Symfony\Component\ExpressionLanguage\Expression;
|
|
use Symfony\Component\ExpressionLanguage\ExpressionLanguage;
|
|
use Symfony\Component\ExpressionLanguage\SyntaxError;
|
|
use Symfony\Component\PropertyAccess\PropertyAccess;
|
|
|
|
/**
|
|
* Automatically update ElasticSearch based on changes to the Doctrine source
|
|
* data. One listener is generated for each Doctrine entity / ElasticSearch type.
|
|
*/
|
|
class Listener implements EventSubscriber
|
|
{
|
|
/**
|
|
* Object persister
|
|
*
|
|
* @var ObjectPersister
|
|
*/
|
|
protected $objectPersister;
|
|
|
|
/**
|
|
* Class of the domain model
|
|
*
|
|
* @var string
|
|
*/
|
|
protected $objectClass;
|
|
|
|
/**
|
|
* List of subscribed events
|
|
*
|
|
* @var array
|
|
*/
|
|
protected $events;
|
|
|
|
/**
|
|
* Name of domain model field used as the ES identifier
|
|
*
|
|
* @var string
|
|
*/
|
|
protected $esIdentifierField;
|
|
|
|
/**
|
|
* Callback for determining if an object should be indexed
|
|
*
|
|
* @var mixed
|
|
*/
|
|
protected $isIndexableCallback;
|
|
|
|
/**
|
|
* Objects scheduled for insertion and replacement
|
|
*/
|
|
public $scheduledForInsertion = array();
|
|
public $scheduledForUpdate = array();
|
|
|
|
/**
|
|
* IDs of objects scheduled for removal
|
|
*/
|
|
public $scheduledForDeletion = array();
|
|
|
|
/**
|
|
* An instance of ExpressionLanguage
|
|
*
|
|
* @var ExpressionLanguage
|
|
*/
|
|
protected $expressionLanguage;
|
|
|
|
/**
|
|
* PropertyAccessor instance
|
|
*
|
|
* @var PropertyAccessorInterface
|
|
*/
|
|
protected $propertyAccessor;
|
|
|
|
/**
|
|
* Constructor.
|
|
*
|
|
* @param ObjectPersisterInterface $objectPersister
|
|
* @param string $objectClass
|
|
* @param array $events
|
|
* @param string $esIdentifierField
|
|
*/
|
|
public function __construct(ObjectPersisterInterface $objectPersister, $objectClass, array $events, $esIdentifierField = 'id', $logger = null)
|
|
{
|
|
$this->objectPersister = $objectPersister;
|
|
$this->objectClass = $objectClass;
|
|
$this->events = $events;
|
|
$this->esIdentifierField = $esIdentifierField;
|
|
|
|
if ($logger) {
|
|
$this->objectPersister->setLogger($logger);
|
|
}
|
|
|
|
$this->propertyAccessor = PropertyAccess::createPropertyAccessor();
|
|
}
|
|
|
|
/**
|
|
* @see Doctrine\Common\EventSubscriber::getSubscribedEvents()
|
|
*/
|
|
public function getSubscribedEvents()
|
|
{
|
|
return $this->events;
|
|
}
|
|
|
|
/**
|
|
* Set the callback for determining object index eligibility.
|
|
*
|
|
* If callback is a string, it must be public method on the object class
|
|
* that expects no arguments and returns a boolean. Otherwise, the callback
|
|
* should expect the object for consideration as its only argument and
|
|
* return a boolean.
|
|
*
|
|
* @param callback $callback
|
|
* @throws \RuntimeException if the callback is not callable
|
|
*/
|
|
public function setIsIndexableCallback($callback)
|
|
{
|
|
if (is_string($callback)) {
|
|
if (!is_callable(array($this->objectClass, $callback))) {
|
|
if (false !== ($expression = $this->getExpressionLanguage())) {
|
|
$callback = new Expression($callback);
|
|
try {
|
|
$expression->compile($callback, array($this->getExpressionVar()));
|
|
} catch (SyntaxError $e) {
|
|
throw new \RuntimeException(sprintf('Indexable callback %s::%s() is not callable or a valid expression.', $this->objectClass, $callback), 0, $e);
|
|
}
|
|
} else {
|
|
throw new \RuntimeException(sprintf('Indexable callback %s::%s() is not callable.', $this->objectClass, $callback));
|
|
}
|
|
}
|
|
} elseif (!is_callable($callback)) {
|
|
if (is_array($callback)) {
|
|
list($class, $method) = $callback + array(null, null);
|
|
if (is_object($class)) {
|
|
$class = get_class($class);
|
|
}
|
|
|
|
if ($class && $method) {
|
|
throw new \RuntimeException(sprintf('Indexable callback %s::%s() is not callable.', $class, $method));
|
|
}
|
|
}
|
|
throw new \RuntimeException('Indexable callback is not callable.');
|
|
}
|
|
|
|
$this->isIndexableCallback = $callback;
|
|
}
|
|
|
|
/**
|
|
* Return whether the object is indexable with respect to the callback.
|
|
*
|
|
* @param object $object
|
|
* @return boolean
|
|
*/
|
|
protected function isObjectIndexable($object)
|
|
{
|
|
if (!$this->isIndexableCallback) {
|
|
return true;
|
|
}
|
|
|
|
if ($this->isIndexableCallback instanceof Expression) {
|
|
return $this->getExpressionLanguage()->evaluate($this->isIndexableCallback, array($this->getExpressionVar($object) => $object));
|
|
}
|
|
|
|
return is_string($this->isIndexableCallback)
|
|
? call_user_func(array($object, $this->isIndexableCallback))
|
|
: call_user_func($this->isIndexableCallback, $object);
|
|
}
|
|
|
|
/**
|
|
* @param mixed $object
|
|
* @return string
|
|
*/
|
|
private function getExpressionVar($object = null)
|
|
{
|
|
$class = $object ?: $this->objectClass;
|
|
$ref = new \ReflectionClass($class);
|
|
|
|
return strtolower($ref->getShortName());
|
|
}
|
|
|
|
/**
|
|
* Provides unified method for retrieving a doctrine object from an EventArgs instance
|
|
*
|
|
* @param EventArgs $eventArgs
|
|
* @return object Entity | Document
|
|
* @throws \RuntimeException if no valid getter is found.
|
|
*/
|
|
private function getDoctrineObject(EventArgs $eventArgs)
|
|
{
|
|
if (method_exists($eventArgs, 'getObject')) {
|
|
return $eventArgs->getObject();
|
|
} elseif (method_exists($eventArgs, 'getEntity')) {
|
|
return $eventArgs->getEntity();
|
|
} elseif (method_exists($eventArgs, 'getDocument')) {
|
|
return $eventArgs->getDocument();
|
|
}
|
|
|
|
throw new \RuntimeException('Unable to retrieve object from EventArgs.');
|
|
}
|
|
|
|
/**
|
|
* @return bool|ExpressionLanguage
|
|
*/
|
|
private function getExpressionLanguage()
|
|
{
|
|
if (null === $this->expressionLanguage) {
|
|
if (!class_exists('Symfony\Component\ExpressionLanguage\ExpressionLanguage')) {
|
|
return false;
|
|
}
|
|
|
|
$this->expressionLanguage = new ExpressionLanguage();
|
|
}
|
|
|
|
return $this->expressionLanguage;
|
|
}
|
|
|
|
public function postPersist(EventArgs $eventArgs)
|
|
{
|
|
$entity = $this->getDoctrineObject($eventArgs);
|
|
|
|
if ($entity instanceof $this->objectClass && $this->isObjectIndexable($entity)) {
|
|
$this->scheduledForInsertion[] = $entity;
|
|
}
|
|
}
|
|
|
|
public function postUpdate(EventArgs $eventArgs)
|
|
{
|
|
$entity = $this->getDoctrineObject($eventArgs);
|
|
|
|
if ($entity instanceof $this->objectClass) {
|
|
if ($this->isObjectIndexable($entity)) {
|
|
$this->scheduledForUpdate[] = $entity;
|
|
} else {
|
|
// Delete if no longer indexable
|
|
$this->scheduleForDeletion($entity);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Delete objects preRemove instead of postRemove so that we have access to the id. Because this is called
|
|
* preRemove, first check that the entity is managed by Doctrine
|
|
*/
|
|
public function preRemove(EventArgs $eventArgs)
|
|
{
|
|
$entity = $this->getDoctrineObject($eventArgs);
|
|
|
|
if ($entity instanceof $this->objectClass) {
|
|
$this->scheduleForDeletion($entity);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Persist scheduled objects to ElasticSearch
|
|
*/
|
|
private function persistScheduled()
|
|
{
|
|
if (count($this->scheduledForInsertion)) {
|
|
$this->objectPersister->insertMany($this->scheduledForInsertion);
|
|
}
|
|
if (count($this->scheduledForUpdate)) {
|
|
$this->objectPersister->replaceMany($this->scheduledForUpdate);
|
|
}
|
|
if (count($this->scheduledForDeletion)) {
|
|
$this->objectPersister->deleteManyByIdentifiers($this->scheduledForDeletion);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Iterate through scheduled actions before flushing to emulate 2.x behavior. Note that the ElasticSearch index
|
|
* will fall out of sync with the source data in the event of a crash during flush.
|
|
*/
|
|
public function preFlush(EventArgs $eventArgs)
|
|
{
|
|
$this->persistScheduled();
|
|
}
|
|
|
|
/**
|
|
* Iterating through scheduled actions *after* flushing ensures that the ElasticSearch index will be affected
|
|
* only if the query is successful
|
|
*/
|
|
public function postFlush(EventArgs $eventArgs)
|
|
{
|
|
$this->persistScheduled();
|
|
}
|
|
|
|
/**
|
|
* Record the specified identifier to delete. Do not need to entire object.
|
|
* @param mixed $object
|
|
* @return mixed
|
|
*/
|
|
protected function scheduleForDeletion($object)
|
|
{
|
|
if ($identifierValue = $this->propertyAccessor->getValue($object, $this->esIdentifierField)) {
|
|
$this->scheduledForDeletion[] = $identifierValue;
|
|
}
|
|
}
|
|
}
|