Symfony2 FOSElasticaBundle update index for all en

Question

I'm using FOSElasticaBundle and Doctrine in my project, and my code works for the selective index update using the Doctrine lifecycle events. The issue I come up against is if I an update a related entity separately.

For example a person may be related to a company through a manytomany relationship. If I update the company name through company entity directly, then indexes for the person related to the company will be out of date and still relate to the company's old name.

I'm a bit lost as to how to handle this, does anyone have any suggestions? Do I have to rely on a scheduled index update and cope with inaccurate index data in the mean time, or is there a way I can call an update for entities related to the entity that has been updated.

I am relying on JMSSerializer groups to establish the mappings. I appreciate this might not be the best way to do things in the longterm.

Answer 1

I've had the same problem. It seems my installation (Symfony 2.5.4 and FOSElastica 3.0.4) differs quite a bit from yours though. Therefore, there were some problems to get the code working. I'm posting my solution, because it may be useful for other developers out there.

The Listener isn't in FOS\ElasticaBundle\Doctrine\ORM\, but in FOS\ElasticaBundle\Doctrine. So you'll have to use that one. Also I had to use Doctrine\Common\EventArgs instead of Doctrine\ORM\Event\LifecycleEventArgs, 'cause otherwise my own postUpdate-method wasn't compatible with the one in the BaseListener.

In my app, a course (seminar) can have a lot of sessions, but in this project, elastica will only be using those sessions. The app needs to know some details of the course that is related to the session of course. So, here's my code:

In config.yml my elastica bundle config looks like this:

fos_elastica:
    clients:
        default: { host: localhost, port: 9200 }
    indexes:
        courses:
            index_name: courses
            types:
                session:
                    mappings:
                        id: ~
                        name: ~
                        course:
                            type: "nested"
                            properties:
                                id: ~
                                name: ~

A little further, still in config.yml

services:
     # some other services here

     fos_elastica.listener.courses.course:
         class: XXX\CourseBundle\EventListener\ElasticaCourseListener
         arguments:
             - @fos_elastica.object_persister.courses.course
             - ['postPersist', 'postUpdate', 'preRemove']
             - @fos_elastica.indexable
         calls:
             - [ setContainer, ['@service_container', @fos_elastica.object_persister.courses.session ] ]
         tags:
             - { name: 'doctrine.event_subscriber' }

My own listener (XXX\CourseBundle\EventListener\ElasticaCourseListener) then looks like this:

<?php

namespace XXX\CourseBundle\EventListener;

use Doctrine\Common\EventArgs;
use FOS\ElasticaBundle\Doctrine\Listener as BaseListener;
use FOS\ElasticaBundle\Persister\ObjectPersister;
use Symfony\Component\DependencyInjection\ContainerInterface;
use XXX\CourseBundle\Entity\Course;

class ElasticaCourseListener extends BaseListener
{
    private $container;
    private $objectPersisterSession;

    public function setContainer(ContainerInterface $container, ObjectPersister $objectPersisterSession)
    {
        $this->container = $container;
        $this->objectPersisterSession = $objectPersisterSession;
    }

    public function postUpdate(EventArgs $args)
    {
        $entity = $args->getEntity();

        if ($entity instanceof Course) {
            $this->scheduledForUpdate[] = $entity;
            foreach ($entity->getSessions() as $session) {
                $this->objectPersisterSession->replaceOne($session);
            }
        }
    }
}

Now, when I update a course, it will be updated as a nested object in ElasticSearch ;-)

Answer 2

I'm using FosElastica 3.1.0 and I have tried the solution provided by Julien Rm without success :-(

After many days of research, I finally found the solution here

$persister = $this->get('fos_elastica.object_persister.jaiuneidee.post');
$persister->insertOne($post);

Hope this help !

Answer 3

with all comments and my research, I made a generic Gist for auto index child objects with fosElastica:

https://gist.github.com/Nightbr/ddb586394d95877dde8ed7445c51d973

In fact, I override the default Listener from FOSElastica and I add the function updateRelations($entity). We will search all relations linked to the $entity and if there are indexed in ES (the ES type exists) it will update the related documents.

If anyone want to look at it and make any improvement it would be great! ^^

Thanks in advance

Answer 4

With the BC Break #729 of FosElastica 3.1.0, things have changed and the code above wasn't working :

BC BREAK: Removed Doctrine\Listener#getSubscribedEvents. The container configuration now configures tags with the methods to call to avoid loading this class on every request where doctrine is active. #729

For those who are trying to make it work with FOSElastica 3.1.X here is how I did manage to make a nested objected to be indexed into his parent into Elastic Search when persisting/updating/removing a nested entity :

Define the service listener :

fos_elastica.listener.entity.nested:
    class: XX\CoreBundle\EventListener\EventSubscriber\ElasticaNestedListener
    arguments:
        - @fos_elastica.object_persister.app.entityname
        - @fos_elastica.indexable
        - {"indexName" : "app", "typeName": "entityname"}
    tags:
        - { name: 'doctrine.event_subscriber' }

Create the listener :

<?php
class ElasticaNestedListener implements EventSubscriber
{ // some indentations missing!

public function getSubscribedEvents()
{
    return array(
        'postPersist',
        'preRemove',
        'postUpdate',
        'preFlush',
        'postFlush',
    );
}

/**
 * Object persister.
 *
 * @var ObjectPersisterInterface
 */
protected $objectPersister;

/**
 * Configuration for the listener.
 *
 * @var array
 */
private $config;

/**
 * Objects scheduled for insertion.
 *
 * @var array
 */
public $scheduledForInsertion = array();

/**
 * Objects scheduled to be updated or removed.
 *
 * @var array
 */
public $scheduledForUpdate = array();

/**
 * IDs of objects scheduled for removal.
 *
 * @var array
 */
public $scheduledForDeletion = array();

/**
 * PropertyAccessor instance.
 *
 * @var PropertyAccessorInterface
 */
protected $propertyAccessor;

/**
 * @var IndexableInterface
 */
private $indexable;

/**
 * Constructor.
 *
 * @param ObjectPersisterInterface $objectPersister
 * @param IndexableInterface       $indexable
 * @param array                    $config
 * @param LoggerInterface          $logger
 */
public function __construct(
    ObjectPersisterInterface $objectPersister,
    IndexableInterface $indexable,
    array $config = array(),
    LoggerInterface $logger = null
) {
    $this->config = array_merge(array(
            'identifier' => 'id',
        ), $config);
    $this->indexable = $indexable;
    $this->objectPersister = $objectPersister;
    $this->propertyAccessor = PropertyAccess::createPropertyAccessor();

    if ($logger && $this->objectPersister instanceof ObjectPersister) {
        $this->objectPersister->setLogger($logger);
    }
}



/**
 * Looks for objects being updated that should be indexed or removed from the index.
 *
 * @param LifecycleEventArgs $eventArgs
 */
public function postUpdate(LifecycleEventArgs $eventArgs)
{
    $entity = $eventArgs->getObject();

    if ($entity instanceof EntityName) {

        $question = $entity->getParent();
        if ($this->objectPersister->handlesObject($question)) {
            if ($this->isObjectIndexable($question)) {
                $this->scheduledForUpdate[] = $question;
            } else {
                // Delete if no longer indexable
                $this->scheduleForDeletion($question);
            }
        }
    }


}


public function postPersist(LifecycleEventArgs $eventArgs)
{
    $entity = $eventArgs->getObject();

    if ($entity instanceof EntityName) {
        $question = $entity->getParent();
        if ($this->objectPersister->handlesObject($question)) {
            if ($this->isObjectIndexable($question)) {
                $this->scheduledForUpdate[] = $question;
            } else {
                // Delete if no longer indexable
                $this->scheduleForDeletion($question);
            }
        }
    }


}


/**
 * 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.
 *
 * @param LifecycleEventArgs $eventArgs
 */
public function preRemove(LifecycleEventArgs $eventArgs)
{
    $entity = $eventArgs->getObject();

    if ($this->objectPersister->handlesObject($entity)) {
        $this->scheduleForDeletion($entity);
    }
}

/**
 * Persist scheduled objects to ElasticSearch
 * After persisting, clear the scheduled queue to prevent multiple data updates when using multiple flush calls.
 */
private function persistScheduled()
{
    if (count($this->scheduledForInsertion)) {
        $this->objectPersister->insertMany($this->scheduledForInsertion);
        $this->scheduledForInsertion = array();
    }
    if (count($this->scheduledForUpdate)) {
        $this->objectPersister->replaceMany($this->scheduledForUpdate);
        $this->scheduledForUpdate = array();
    }
    if (count($this->scheduledForDeletion)) {
        $this->objectPersister->deleteManyByIdentifiers($this->scheduledForDeletion);
        $this->scheduledForDeletion = array();
    }
}

/**
 * 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.
 *
 * This method is only called in legacy configurations of the listener.
 *
 * @deprecated This method should only be called in applications that depend
 *             on the behaviour that entities are indexed regardless of if a
 *             flush is successful.
 */
public function preFlush()
{
    $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()
{
    $this->persistScheduled();
}

/**
 * Record the specified identifier to delete. Do not need to entire object.
 *
 * @param object $object
 */
private function scheduleForDeletion($object)
{
    if ($identifierValue = $this->propertyAccessor->getValue($object, $this->config['identifier'])) {
        $this->scheduledForDeletion[] = $identifierValue;
    }
}

/**
 * Checks if the object is indexable or not.
 *
 * @param object $object
 *
 * @return bool
 */
private function isObjectIndexable($object)
{
    return $this->indexable->isObjectIndexable(
        $this->config['indexName'],
        $this->config['typeName'],
        $object
    );
}
}

EntityName could be a Comment and getParent() could be the Article who owns this comment ...

Hope this help !

Answer 5

I think I've found the solution on this page https://groups.google.com/forum/#!topic/elastica-php-client/WTONX-zBTI4 Thanks Cassiano

Basically you need to extend the FOS\ElasticaBundle\Doctrine\ORM\Listener so you can look for related entities and then update their index as well.

class CompanyListener extends BaseListener
{

    /** @var \Symfony\Component\DependencyInjection\ContainerInterface */
    private $container;

    public function setContainer(\Symfony\Component\DependencyInjection\ContainerInterface $container) {
        $this->container = $container;
    }

    protected function initialiseJob() {
        $this->objectPersisterJob = $this->container->get('fos_elastica.object_persister.application.job');
        $this->em = $this->container->get('doctrine')->getEntityManager(); //maybe move this to postUpdate function so it can be used for all
    }

    /**
     * @param \Doctrine\ORM\Event\LifecycleEventArgs $eventArgs
     */
    public function postUpdate(LifecycleEventArgs $eventArgs)
    {
        /** @var $entity Story */
        $entity = $eventArgs->getEntity();

        if ($entity instanceof $this->objectClass) {
            if ($this->isObjectIndexable($entity)) {
                $this->objectPersister->replaceOne($entity);
                $this->initialiseJob();
                foreach ($entity->getJobOpenings() as $job) {
                    $this->objectPersisterJob->replaceOne($job);
                }
            } else {
                $this->scheduleForRemoval($entity, $eventArgs->getEntityManager());
                $this->removeIfScheduled($entity);
            }
        }
    }

    public function preRemove(\Doctrine\Common\EventArgs $eventArgs)
    {
        $entity = $eventArgs->getEntity();

        if ($entity instanceof $this->objectClass) {

            $this->scheduleForDeletion($entity);
            $this->initialiseJob();
            foreach ($entity->getJobOpenings() as $job) {
                $this->objectPersisterJob->replaceOne($job);
            }
        }
    }


}

and your services defined as below

fos_elastica.listener.application.company:
    class: 'xxx\RMSBundle\EventListener\CompanyListener'
    arguments:
        - '@fos_elastica.object_persister.application.company'
        - 'xxx\RMSBundle\Entity\Company'
        - ['postPersist', 'postUpdate', 'postRemove', 'preRemove']
        - id
    calls:
        - [ setContainer, [ '@service_container' ] ]
    tags:
        - { name: 'doctrine.event_subscriber' }

this will then update indexes for both :-)

Answer 6

Sorry, i can not comment under your answer but something is missing in the solution. You have to override preRemove too.

public function preRemove(\Doctrine\Common\EventArgs $eventArgs)
{
    $entity = $eventArgs->getEntity();



    if ($entity instanceof $this->objectClass) {

        $this->scheduleForDeletion($entity);
        $this->initialiseJob();
        foreach ($entity->getJobOpenings() as $job) {
                $this->objectPersisterJob->replaceOne($job);
            }
    }
}