<commit_before><?php
/**
 * Date: 25/01/16
 * Time: 13:27
 */

namespace Broadway\ReadModel\ElasticSearch;


use Assert\Assertion;
use Broadway\ReadModel\ReadModelInterface;
use Broadway\Serializer\SerializerInterface;
use Elasticsearch\Client;
use Elasticsearch\Common\Exceptions\Conflict409Exception;
use Elasticsearch\Common\Exceptions\Missing404Exception;

class AdvancedElasticSearchRepository extends ElasticSearchRepository
{
    private $client;
    private $serializer;
    private $index;
    private $class;
    private $notAnalyzedFields;

    private $models = [];
    private $versions = [];

    /**
     * @param string $index
     * @param string $class
     * @param array $notAnalyzedFields = array
     */
    public function __construct(
        Client $client,
        SerializerInterface $serializer,
        $index,
        $class,
        array $notAnalyzedFields = array(),
        $environment
    ) {
        parent::__construct($client, $serializer, $index, $class, $notAnalyzedFields, $environment);

        $this->client = $client;
        $this->serializer = $serializer;
        $this->index = $environment.'_'.$index;
        $this->class = $class;
        $this->notAnalyzedFields = $notAnalyzedFields;
        $this->models = [];
        $this->versions = [];
    }

    /**
     * @param ReadModelInterface $data
     * @param bool $flush
     * @throws Conflict409Exception
     */
    public function save(ReadModelInterface $data, $flush = true)
    {
        Assertion::isInstanceOf($data, $this->class);

        $this->models[$data->getId()] = $data;

        if ($flush) {
            $this->flush($data);
        }
    }

    /**
     * Actually persist in memory readmodels to elasticsearch
     * @throws Conflict409Exception
     */
    protected function flush(ReadModelInterface $data)
    {
        $serializedReadModel = $this->serializer->serialize($data);

        $params = [
            'index' => $this->index,
            'type' => $serializedReadModel['class'],
            'id' => $data->getId(),
            'body' => $serializedReadModel['payload'],
            "refresh" => "true",
        ];

        if (isset($this->versions[$data->getId()])) {
            $params['version'] = $this->versions[$data->getId()];
        } else {
            $this->versions[$data->getId()] = 0;
        }

        try {
            $this->client->index($params);
        } catch (Conflict409Exception $e) {
            unset($this->versions[$data->getId()]); // make sure to remove in memory data because it's not up to date
            unset($this->models[$data->getId()]);

            throw $e;// Enable retrying upstream
        }

        $this->versions[$data->getId()] += 1;
    }

    /**
     *  Persist all in memory readmodels
     */
    public function flushAll()
    {
        foreach ($this->models as $model) {
            $this->flush($model);
        }
    }

    public function remove($id)
    {
        parent::remove($id);

        if (array_key_exists((string)($id), $this->models)) {
            unset($this->models[(string)$id]);
            unset($this->versions[(string)$id]);
        }
    }


    /**
     * @param string $id
     * @return mixed|null
     */
    public function find($id)
    {
        if (array_key_exists((string)($id), $this->models)) {
            return $this->models[(string)$id];<commit_msg>Only persist models that have actually been changed<commit_after><?php
/**
 * Date: 25/01/16
 * Time: 13:27
 */

namespace Broadway\ReadModel\ElasticSearch;


use Assert\Assertion;
use Broadway\ReadModel\ReadModelInterface;
use Broadway\Serializer\SerializerInterface;
use Elasticsearch\Client;
use Elasticsearch\Common\Exceptions\Conflict409Exception;
use Elasticsearch\Common\Exceptions\Missing404Exception;

class AdvancedElasticSearchRepository extends ElasticSearchRepository
{
    private $client;
    private $serializer;
    private $index;
    private $class;
    private $notAnalyzedFields;

    private $models = [];
    private $versions = [];
    private $changed = [];

    /**
     * @param string $index
     * @param string $class
     * @param array $notAnalyzedFields = array
     */
    public function __construct(
        Client $client,
        SerializerInterface $serializer,
        $index,
        $class,
        array $notAnalyzedFields = array(),
        $environment
    ) {
        parent::__construct($client, $serializer, $index, $class, $notAnalyzedFields, $environment);

        $this->client = $client;
        $this->serializer = $serializer;
        $this->index = $environment.'_'.$index;
        $this->class = $class;
        $this->notAnalyzedFields = $notAnalyzedFields;
        $this->models = [];
        $this->versions = [];
    }

    /**
     * @param ReadModelInterface $data
     * @param bool $flush
     * @throws Conflict409Exception
     */
    public function save(ReadModelInterface $data, $flush = true)
    {
        Assertion::isInstanceOf($data, $this->class);

        $this->models[$data->getId()] = $data;
        $this->changed[$data->getId()] = true;

        if ($flush) {
            $this->flush($data);
        }
    }

    /**
     * Actually persist in memory readmodels to elasticsearch
     * @throws Conflict409Exception
     */
    protected function flush(ReadModelInterface $data)
    {
        $serializedReadModel = $this->serializer->serialize($data);

        $params = [
            'index' => $this->index,
            'type' => $serializedReadModel['class'],
            'id' => $data->getId(),
            'body' => $serializedReadModel['payload'],
            "refresh" => "true",
        ];

        if (isset($this->versions[$data->getId()])) {
            $params['version'] = $this->versions[$data->getId()];
        } else {
            $this->versions[$data->getId()] = 0;
        }

        try {
            $this->client->index($params);
        } catch (Conflict409Exception $e) {
            unset($this->versions[$data->getId()]); // make sure to remove in memory data because it's not up to date
            unset($this->models[$data->getId()]);

            throw $e;// Enable retrying upstream
        }

        $this->versions[$data->getId()] += 1;
    }

    /**
     *  Persist all in memory readmodels
     */
    public function flushAll()
    {
        foreach ($this->changed as $modelKey => $value) {
            $this->flush($this->models[$modelKey]);
        }
    }

    public function remove($id)
    {
        parent::remove($id);

        if (array_key_exists((string)($id), $this->models)) {
            unset($this->models[(string)$id]);
            unset($this->versions[(string)$id]);
        }
    }


    /**
     * @param string $id
     * @return mixed|null
     */
    public function find($id)
    {
        if (array_key_exists((string)($id), $this->models)) {
            return $this->models[(string)$id];
{'fraction_non_alphanumeric': 0.14988459319099826, 'fraction_numerical': 0.007789959607616849, 'mean_word_length': 2.310888252148997, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 89, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5045720', 'n_tokens_mistral': 2005, 'n_tokens_neox': 1955, 'n_words': 399}