Manager::persist() - Code Metrics - Inspection of "implemented retrievement of id after commit()" - ongr-io/ElasticsearchBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#589)

unknown

created 2016-03-16 09:06 UTC

Manager::persist() B

↳ Parent: Manager

Complexity

Conditions	6
Paths	6

Size

Total Lines	21
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	1	Features	1

Metric	Value
c	3
b	1
f	1
dl	0
loc	21
rs	8.7624
cc	6
eloc	13
nc	6
nop	1

<?php

/*
 * This file is part of the ONGR package.
 *
 * (c) NFQ Technologies UAB <[email protected]>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace ONGR\ElasticsearchBundle\Service;

use Elasticsearch\Client;
use Elasticsearch\Common\Exceptions\Forbidden403Exception;
use Elasticsearch\Common\Exceptions\Missing404Exception;
use ONGR\ElasticsearchBundle\Mapping\MetadataCollector;
use ONGR\ElasticsearchBundle\Result\AbstractResultsIterator;
use ONGR\ElasticsearchBundle\Result\Converter;
use ONGR\ElasticsearchBundle\Result\DocumentIterator;
use ONGR\ElasticsearchBundle\Result\RawIterator;
use ONGR\ElasticsearchBundle\Result\Result;
use ONGR\ElasticsearchDSL\Search;

/**
 * Manager class.
 */
class Manager
{
    /**
     * @var string Manager name
     */
    private $name;

    /**
     * @var array Manager configuration
     */
    private $config = [];

    /**
     * @var Client
     */
    private $client;

    /**
     * @var Converter
     */
    private $converter;

    /**
     * @var bool
     */
    private $readOnly;

    /**
     * @var array Container for bulk queries
     */
    private $bulkQueries = [];

    /**
     * @var array Holder for consistency, refresh and replication parameters
     */
    private $bulkParams = [];

    /**
     * @var array
     */
    private $indexSettings;

    /**
     * @var MetadataCollector
     */
    private $metadataCollector;

    /**
     * After commit to make data available the refresh or flush operation is needed
     * so one of those methods has to be defined, the default is refresh.
     *
     * @var string
     */
    private $commitMode = 'refresh';

    /**
     * Reference to the persisted objects that need to be given an ID that returns from
     * elasticsearch
     *
     * @var array
     */
    private $persistedObjects = [];

    /**
     * The size that defines after how much document inserts call commit function.
     *
     * @var int
     */
    private $bulkCommitSize = 100;

    /**
     * Container to count how many documents was passed to the bulk query.
     *
     * @var int
     */
    private $bulkCount = 0;

    /**
     * @var Repository[] Repository local cache
     */
    private $repositories;

    /**
     * @param string            $name              Manager name
     * @param array             $config            Manager configuration
     * @param Client            $client
     * @param array             $indexSettings
     * @param MetadataCollector $metadataCollector
     * @param Converter         $converter
     */
    public function __construct(
        $name,
        array $config,
        $client,
        array $indexSettings,
        $metadataCollector,
        $converter
    ) {
        $this->name = $name;
        $this->config = $config;
        $this->client = $client;
        $this->indexSettings = $indexSettings;
        $this->metadataCollector = $metadataCollector;
        $this->converter = $converter;

        $this->setReadOnly($config['readonly']);
    }

    /**
     * Returns Elasticsearch connection.
     *
     * @return Client
     */
    public function getClient()
    {
        return $this->client;
    }

    /**
     * @return string
     */
    public function getName()
    {
        return $this->name;
    }

    /**
     * @return array
     */
    public function getConfig()
    {
        return $this->config;
    }

    /**
     * Returns repository by document class.
     *
     * @param string $className FQCN or string in Bundle:Document format
     *
     * @return Repository
     */
    public function getRepository($className)
    {
        if (!is_string($className)) {
            throw new \InvalidArgumentException('Document class must be a string.');
        }

        $namespace = $this->getMetadataCollector()->getClassName($className);

        if (isset($this->repositories[$namespace])) {
            return $this->repositories[$namespace];
        }

        $repository = $this->createRepository($namespace);
        $this->repositories[$namespace] = $repository;

        return $repository;
    }

    /**
     * @return MetadataCollector
     */
    public function getMetadataCollector()
    {
        return $this->metadataCollector;
    }

    /**
     * @return Converter
     */
    public function getConverter()
    {
        return $this->converter;
    }

    /**
     * @return string
     */
    public function getCommitMode()
    {
        return $this->commitMode;
    }

    /**
     * @param string $commitMode
     */
    public function setCommitMode($commitMode)
    {
        if ($commitMode === 'refresh' || $commitMode === 'flush' || $commitMode === 'none') {
            $this->commitMode = $commitMode;
        } else {
            throw new \LogicException('The commit method must be either refresh, flush or none.');
        }
    }

    /**
     * @return int
     */
    public function getBulkCommitSize()
    {
        return $this->bulkCommitSize;
    }

    /**
     * @param int $bulkCommitSize
     */
    public function setBulkCommitSize($bulkCommitSize)
    {
        $this->bulkCommitSize = $bulkCommitSize;
    }

    /**
     * Creates a repository.
     *
     * @param string $className
     *
     * @return Repository
     */
    private function createRepository($className)
    {
        return new Repository($this, $className);
    }

    /**
     * Executes search query in the index.
     *
     * @param array $types             List of types to search in.
     * @param array $query             Query to execute.
     * @param array $queryStringParams Query parameters.
     *
     * @return array
     */
    public function search(array $types, array $query, array $queryStringParams = [])
    {
        $params = [];
        $params['index'] = $this->getIndexName();
        $params['type'] = implode(',', $types);
        $params['body'] = $query;

        if (!empty($queryStringParams)) {
            $params = array_merge($queryStringParams, $params);
        }

        return $this->client->search($params);
    }

    /**
     * Adds document to next flush.
     *
     * @param object $document
     */
    public function persist($document)
    {
        $documentArray = $this->converter->convertToArray($document);
        $type = $this->getMetadataCollector()->getDocumentType(get_class($document));

        $this->bulk('index', $type, $documentArray);

        if ($id_field_info = $this->getIdFieldInfo($document)) {
            if ($id_field_info['property_type'] == 'public') {
                $id_field = $id_field_info['id_property'];
                if (!isset($document->$id_field)) {
                    $this->persistedObjects[] = $document;
                }
            } elseif (isset($id_field_info['getter'])) {
                $getter = $id_field_info['getter'];
                if ($document->$getter() == null) {
                    $this->persistedObjects[] = $document;
                }
            }
        }
    }

    /**
     * Adds document for removal.
     *
     * @param object $document
     */
    public function remove($document)
    {
        $data = $this->converter->convertToArray($document, [], ['_id']);

        if (!isset($data['_id'])) {
            throw new \LogicException(
                'In order to use remove() method document class must have property with @Id annotation.'
            );
        }

        $type = $this->getMetadataCollector()->getDocumentType(get_class($document));

        $this->bulk('delete', $type, ['_id' => $data['_id']]);
    }

    /**
     * Flushes elasticsearch index.
     *
     * @param array $params
     *
     * @return array
     */
    public function flush(array $params = [])
    {
        return $this->client->indices()->flush($params);
    }

    /**
     * Refreshes elasticsearch index.
     *
     * @param array $params
     *
     * @return array
     */
    public function refresh(array $params = [])
    {
        return $this->client->indices()->refresh($params);
    }

    /**
     * Gets Id field of the document
     *
     * @param object $document
     *
     * @return mixed
     */
    public function getIdFieldInfo($document)
    {
        $class = get_class($document);
        $mapping = $this->metadataCollector->getMapping($class);
        $response = [];

        if (isset($mapping['aliases']['_id'])) {
            $mapping = $mapping['aliases']['_id'];
            $response['id_property'] = $mapping['propertyName'];
            $response['property_type'] = 'private';

            if ($mapping['propertyType'] == 'public') {
                $response['property_type'] = 'public';
            } elseif (isset($mapping['methods']['setter']) && isset($mapping['methods']['getter'])) {
                $response['setter'] = $mapping['methods']['setter'];
                $response['getter'] = $mapping['methods']['getter'];
            }
        } else {
            return false;
        }
        return $response;
    }

    /**
     * Adds ids to documents
     *
     * @param array $bulkQueries
     *
     * @param array $bulkResponse
     */
    public function addIds(array $bulkQueries, $bulkResponse = [])
    {
        if (empty($bulkResponse)) {
            $this->persistedObjects = [];
            return;
        }
        $indexing = [];
        foreach ($bulkQueries['body'] as $number => $query) {
            if (isset($query['index']) && !isset($query['index']['_id'])) {
                $indexing[] = $number / 2;
            }
        }
        if (isset($bulkQueries['body'][0]['index'])) {
            if (isset($this->persistedObjects)) {
                $i = 0;
                foreach ($this->persistedObjects as $document) {
                    if ($id_field_info = $this->getIdFieldInfo($document)) {
                        if ($id_field_info['property_type'] == 'public') {
                            $id_field = $id_field_info['id_property'];
                            $document->$id_field = $bulkResponse['items'][$indexing[$i]]['create']['_id'];
                        } elseif ($id_field_info['setter']) {
                            $method = $id_field_info['setter'];
                            $document->$method($bulkResponse['items'][$indexing[$i]]['create']['_id']);
                        }
                    }
                    $i++;
                }
            }
            $this->persistedObjects = [];
        }
    }

    /**
     * Inserts the current query container to the index, used for bulk queries execution.
     *
     * @param array $params Parameters that will be passed to the flush or refresh queries.
     *
     * @return null|array
     */
    public function commit(array $params = [])
    {
        $this->isReadOnly('Commit');

        if (!empty($this->bulkQueries)) {
            $bulkQueries = array_merge($this->bulkQueries, $this->bulkParams);
            $this->bulkQueries = [];

            $bulkResponse = $this->client->bulk($bulkQueries);

            switch ($this->getCommitMode()) {
                case 'flush':
                    $this->flush($params);
                    break;
                case 'refresh':
                    $this->refresh($params);
                    break;
            }

            $this->addIds($bulkQueries, $bulkResponse);
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);

            return $bulkResponse;
        }
        return null;
    }

    /**
     * Adds query to bulk queries container.
     *
     * @param string       $operation One of: index, update, delete, create.
     * @param string|array $type      Elasticsearch type name.
     * @param array        $query     DSL to execute.
     *
     * @throws \InvalidArgumentException
     */
    public function bulk($operation, $type, array $query)
    {
        $this->isReadOnly('Bulk');

        if (!in_array($operation, ['index', 'create', 'update', 'delete'])) {
            throw new \InvalidArgumentException('Wrong bulk operation selected');
        }

        $this->bulkQueries['body'][] = [
            $operation => array_filter(
                [
                    '_index' => $this->getIndexName(),
                    '_type' => $type,
                    '_id' => isset($query['_id']) ? $query['_id'] : null,
                    '_ttl' => isset($query['_ttl']) ? $query['_ttl'] : null,
                    '_parent' => isset($query['_parent']) ? $query['_parent'] : null,
                ]
            ),
        ];
        unset($query['_id'], $query['_ttl'], $query['_parent']);

        switch ($operation) {
            case 'index':
            case 'create':
            case 'update':
                $this->bulkQueries['body'][] = $query;
                break;
            case 'delete':
                // Body for delete operation is not needed to apply.
            default:
                // Do nothing.
                break;
        }

        // We are using counter because there is to difficult to resolve this from bulkQueries array.
        $this->bulkCount++;

        if ($this->bulkCommitSize === $this->bulkCount) {
            $this->commit();
            $this->bulkCount = 0;
        }
    }

    /**
     * Optional setter to change bulk query params.
     *
     * @param array $params Possible keys:
     *                      ['consistency'] = (enum) Explicit write consistency setting for the operation.
     *                      ['refresh']     = (boolean) Refresh the index after performing the operation.
     *                      ['replication'] = (enum) Explicitly set the replication type.
     */
    public function setBulkParams(array $params)
    {
        $this->bulkParams = $params;
    }

    /**
     * Creates fresh elasticsearch index.
     *
     * @param bool $noMapping Determines if mapping should be included.
     *
     * @return array
     */
    public function createIndex($noMapping = false)
    {
        $this->isReadOnly('Create index');

        if ($noMapping) {
            unset($this->indexSettings['body']['mappings']);
        }

        return $this->getClient()->indices()->create($this->indexSettings);
    }

    /**
     * Drops elasticsearch index.
     */
    public function dropIndex()
    {
        $this->isReadOnly('Drop index');

        return $this->getClient()->indices()->delete(['index' => $this->getIndexName()]);
    }

    /**
     * Tries to drop and create fresh elasticsearch index.
     *
     * @param bool $noMapping Determines if mapping should be included.
     *
     * @return array
     */
    public function dropAndCreateIndex($noMapping = false)
    {
        try {
            $this->dropIndex();
        } catch (\Exception $e) {
            // Do nothing, our target is to create new index.
        }

        return $this->createIndex($noMapping);
    }

    /**
     * Checks if connection index is already created.
     *
     * @return bool
     */
    public function indexExists()
    {
        return $this->getClient()->indices()->exists(['index' => $this->getIndexName()]);
    }

    /**
     * Returns index name this connection is attached to.
     *
     * @return string
     */
    public function getIndexName()
    {
        return $this->indexSettings['index'];
    }

    /**
     * Sets index name for this connection.
     *
     * @param string $name
     */
    public function setIndexName($name)
    {
        $this->indexSettings['index'] = $name;
    }

    /**
     * Returns Elasticsearch version number.
     *
     * @return string
     */
    public function getVersionNumber()
    {
        return $this->client->info()['version']['number'];
    }

    /**
     * Clears elasticsearch client cache.
     */
    public function clearCache()
    {
        $this->isReadOnly('Clear cache');

        $this->getClient()->indices()->clearCache(['index' => $this->getIndexName()]);
    }

    /**
     * Set connection to read only state.
     *
     * @param bool $readOnly
     */
    public function setReadOnly($readOnly)
    {
        $this->readOnly = $readOnly;
    }

    /**
     * Checks if connection is read only.
     *
     * @param string $message Error message.
     *
     * @throws Forbidden403Exception
     */
    public function isReadOnly($message = '')
    {
        if ($this->readOnly) {
            throw new Forbidden403Exception("Manager is readonly! {$message} operation is not permitted.");
        }
    }

    /**
     * Returns a single document by ID. Returns NULL if document was not found.
     *
     * @param string $className Document class name or Elasticsearch type name
     * @param string $id        Document ID to find
     *
     * @return object
     */
    public function find($className, $id)
    {
        $type = $this->resolveTypeName($className);

        $params = [
            'index' => $this->getIndexName(),
            'type' => $type,
            'id' => $id,
        ];

        try {
            $result = $this->getClient()->get($params);
        } catch (Missing404Exception $e) {
            return null;
        }

        return $this->getConverter()->convertToDocument($result, $this);
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
    }

    /**
     * Executes given search.
     *
     * @param array  $types
     * @param Search $search
     * @param string $resultsType
     *
     * @return DocumentIterator|RawIterator|array
     */
    public function execute($types, Search $search, $resultsType = Result::RESULTS_OBJECT)
    {
        foreach ($types as &$type) {
            $type = $this->resolveTypeName($type);
        }

        $results = $this->search($types, $search->toArray(), $search->getQueryParams());

        return $this->parseResult($results, $resultsType, $search->getScroll());
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
    }

    /**
     * Parses raw result.
     *
     * @param array  $raw
     * @param string $resultsType
     * @param string $scrollDuration
     *
     * @return DocumentIterator|RawIterator|array
     *
     * @throws \Exception
     */
    private function parseResult($raw, $resultsType, $scrollDuration = null)
    {
        $scrollConfig = [];
        if (isset($raw['_scroll_id'])) {
            $scrollConfig['_scroll_id'] = $raw['_scroll_id'];
            $scrollConfig['duration'] = $scrollDuration;
        }

        switch ($resultsType) {
            case Result::RESULTS_OBJECT:
                return new DocumentIterator($raw, $this, $scrollConfig);
            case Result::RESULTS_ARRAY:
                return $this->convertToNormalizedArray($raw);
            case Result::RESULTS_RAW:
                return $raw;
            case Result::RESULTS_RAW_ITERATOR:
                return new RawIterator($raw, $this, $scrollConfig);
            default:
                throw new \Exception('Wrong results type selected');
        }
    }

    /**
     * Normalizes response array.
     *
     * @param array $data
     *
     * @return array
     */
    private function convertToNormalizedArray($data)
    {
        if (array_key_exists('_source', $data)) {
            return $data['_source'];
        }

        $output = [];

        if (isset($data['hits']['hits'][0]['_source'])) {
            foreach ($data['hits']['hits'] as $item) {
                $output[] = $item['_source'];
            }
        } elseif (isset($data['hits']['hits'][0]['fields'])) {
            foreach ($data['hits']['hits'] as $item) {
                $output[] = array_map('reset', $item['fields']);
            }
        }

        return $output;
    }

    /**
     * Fetches next set of results.
     *
     * @param string $scrollId
     * @param string $scrollDuration
     * @param string $resultsType
     *
     * @return AbstractResultsIterator
     *
     * @throws \Exception
     */
    public function scroll(
        $scrollId,
        $scrollDuration = '5m',
        $resultsType = Result::RESULTS_OBJECT
    ) {
        $results = $this->getClient()->scroll(['scroll_id' => $scrollId, 'scroll' => $scrollDuration]);

        return $this->parseResult($results, $resultsType, $scrollDuration);
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
    }

    /**
     * Clears scroll.
     *
     * @param string $scrollId
     */
    public function clearScroll($scrollId)
    {
        $this->getClient()->clearScroll(['scroll_id' => $scrollId]);
    }

    /**
     * Resolves type name by class name.
     *
     * @param string $className
     *
     * @return string
     */
    private function resolveTypeName($className)
    {
        if (strpos($className, ':') !== false || strpos($className, '\\') !== false) {
            return $this->getMetadataCollector()->getDocumentType($className);
        }

        return $className;
    }
}


1			<?php
2
3			/*
4			* This file is part of the ONGR package.
5			*
6			* (c) NFQ Technologies UAB <[email protected]>
7			*
8			* For the full copyright and license information, please view the LICENSE
9			* file that was distributed with this source code.
10			*/
11
12			namespace ONGR\ElasticsearchBundle\Service;
13
14			use Elasticsearch\Client;
15			use Elasticsearch\Common\Exceptions\Forbidden403Exception;
16			use Elasticsearch\Common\Exceptions\Missing404Exception;
17			use ONGR\ElasticsearchBundle\Mapping\MetadataCollector;
18			use ONGR\ElasticsearchBundle\Result\AbstractResultsIterator;
19			use ONGR\ElasticsearchBundle\Result\Converter;
20			use ONGR\ElasticsearchBundle\Result\DocumentIterator;
21			use ONGR\ElasticsearchBundle\Result\RawIterator;
22			use ONGR\ElasticsearchBundle\Result\Result;
23			use ONGR\ElasticsearchDSL\Search;
24
25			/**
26			* Manager class.
27			*/
28			class Manager
29			{
30			/**
31			* @var string Manager name
32			*/
33			private $name;
34
35			/**
36			* @var array Manager configuration
37			*/
38			private $config = [];
39
40			/**
41			* @var Client
42			*/
43			private $client;
44
45			/**
46			* @var Converter
47			*/
48			private $converter;
49
50			/**
51			* @var bool
52			*/
53			private $readOnly;
54
55			/**
56			* @var array Container for bulk queries
57			*/
58			private $bulkQueries = [];
59
60			/**
61			* @var array Holder for consistency, refresh and replication parameters
62			*/
63			private $bulkParams = [];
64
65			/**
66			* @var array
67			*/
68			private $indexSettings;
69
70			/**
71			* @var MetadataCollector
72			*/
73			private $metadataCollector;
74
75			/**
76			* After commit to make data available the refresh or flush operation is needed
77			* so one of those methods has to be defined, the default is refresh.
78			*
79			* @var string
80			*/
81			private $commitMode = 'refresh';
82
83			/**
84			* Reference to the persisted objects that need to be given an ID that returns from
85			* elasticsearch
86			*
87			* @var array
88			*/
89			private $persistedObjects = [];
90
91			/**
92			* The size that defines after how much document inserts call commit function.
93			*
94			* @var int
95			*/
96			private $bulkCommitSize = 100;
97
98			/**
99			* Container to count how many documents was passed to the bulk query.
100			*
101			* @var int
102			*/
103			private $bulkCount = 0;
104
105			/**
106			* @var Repository[] Repository local cache
107			*/
108			private $repositories;
109
110			/**
111			* @param string $name Manager name
112			* @param array $config Manager configuration
113			* @param Client $client
114			* @param array $indexSettings
115			* @param MetadataCollector $metadataCollector
116			* @param Converter $converter
117			*/
118			public function __construct(
119			$name,
120			array $config,
121			$client,
122			array $indexSettings,
123			$metadataCollector,
124			$converter
125			) {
126			$this->name = $name;
127			$this->config = $config;
128			$this->client = $client;
129			$this->indexSettings = $indexSettings;
130			$this->metadataCollector = $metadataCollector;
131			$this->converter = $converter;
132
133			$this->setReadOnly($config['readonly']);
134			}
135
136			/**
137			* Returns Elasticsearch connection.
138			*
139			* @return Client
140			*/
141			public function getClient()
142			{
143			return $this->client;
144			}
145
146			/**
147			* @return string
148			*/
149			public function getName()
150			{
151			return $this->name;
152			}
153
154			/**
155			* @return array
156			*/
157			public function getConfig()
158			{
159			return $this->config;
160			}
161
162			/**
163			* Returns repository by document class.
164			*
165			* @param string $className FQCN or string in Bundle:Document format
166			*
167			* @return Repository
168			*/
169			public function getRepository($className)
170			{
171			if (!is_string($className)) {
172			throw new \InvalidArgumentException('Document class must be a string.');
173			}
174
175			$namespace = $this->getMetadataCollector()->getClassName($className);
176
177			if (isset($this->repositories[$namespace])) {
178			return $this->repositories[$namespace];
179			}
180
181			$repository = $this->createRepository($namespace);
182			$this->repositories[$namespace] = $repository;
183
184			return $repository;
185			}
186
187			/**
188			* @return MetadataCollector
189			*/
190			public function getMetadataCollector()
191			{
192			return $this->metadataCollector;
193			}
194
195			/**
196			* @return Converter
197			*/
198			public function getConverter()
199			{
200			return $this->converter;
201			}
202
203			/**
204			* @return string
205			*/
206			public function getCommitMode()
207			{
208			return $this->commitMode;
209			}
210
211			/**
212			* @param string $commitMode
213			*/
214			public function setCommitMode($commitMode)
215			{
216			if ($commitMode === 'refresh' \|\| $commitMode === 'flush' \|\| $commitMode === 'none') {
217			$this->commitMode = $commitMode;
218			} else {
219			throw new \LogicException('The commit method must be either refresh, flush or none.');
220			}
221			}
222
223			/**
224			* @return int
225			*/
226			public function getBulkCommitSize()
227			{
228			return $this->bulkCommitSize;
229			}
230
231			/**
232			* @param int $bulkCommitSize
233			*/
234			public function setBulkCommitSize($bulkCommitSize)
235			{
236			$this->bulkCommitSize = $bulkCommitSize;
237			}
238
239			/**
240			* Creates a repository.
241			*
242			* @param string $className
243			*
244			* @return Repository
245			*/
246			private function createRepository($className)
247			{
248			return new Repository($this, $className);
249			}
250
251			/**
252			* Executes search query in the index.
253			*
254			* @param array $types List of types to search in.
255			* @param array $query Query to execute.
256			* @param array $queryStringParams Query parameters.
257			*
258			* @return array
259			*/
260			public function search(array $types, array $query, array $queryStringParams = [])
261			{
262			$params = [];
263			$params['index'] = $this->getIndexName();
264			$params['type'] = implode(',', $types);
265			$params['body'] = $query;
266
267			if (!empty($queryStringParams)) {
268			$params = array_merge($queryStringParams, $params);
269			}
270
271			return $this->client->search($params);
272			}
273
274			/**
275			* Adds document to next flush.
276			*
277			* @param object $document
278			*/
279			public function persist($document)
280			{
281			$documentArray = $this->converter->convertToArray($document);
282			$type = $this->getMetadataCollector()->getDocumentType(get_class($document));
283
284			$this->bulk('index', $type, $documentArray);
285
286			if ($id_field_info = $this->getIdFieldInfo($document)) {
287			if ($id_field_info['property_type'] == 'public') {
288			$id_field = $id_field_info['id_property'];
289			if (!isset($document->$id_field)) {
290			$this->persistedObjects[] = $document;
291			}
292			} elseif (isset($id_field_info['getter'])) {
293			$getter = $id_field_info['getter'];
294			if ($document->$getter() == null) {
295			$this->persistedObjects[] = $document;
296			}
297			}
298			}
299			}
300
301			/**
302			* Adds document for removal.
303			*
304			* @param object $document
305			*/
306			public function remove($document)
307			{
308			$data = $this->converter->convertToArray($document, [], ['_id']);
309
310			if (!isset($data['_id'])) {
311			throw new \LogicException(
312			'In order to use remove() method document class must have property with @Id annotation.'
313			);
314			}
315
316			$type = $this->getMetadataCollector()->getDocumentType(get_class($document));
317
318			$this->bulk('delete', $type, ['_id' => $data['_id']]);
319			}
320
321			/**
322			* Flushes elasticsearch index.
323			*
324			* @param array $params
325			*
326			* @return array
327			*/
328			public function flush(array $params = [])
329			{
330			return $this->client->indices()->flush($params);
331			}
332
333			/**
334			* Refreshes elasticsearch index.
335			*
336			* @param array $params
337			*
338			* @return array
339			*/
340			public function refresh(array $params = [])
341			{
342			return $this->client->indices()->refresh($params);
343			}
344
345			/**
346			* Gets Id field of the document
347			*
348			* @param object $document
349			*
350			* @return mixed
351			*/
352			public function getIdFieldInfo($document)
353			{
354			$class = get_class($document);
355			$mapping = $this->metadataCollector->getMapping($class);
356			$response = [];
357
358			if (isset($mapping['aliases']['_id'])) {
359			$mapping = $mapping['aliases']['_id'];
360			$response['id_property'] = $mapping['propertyName'];
361			$response['property_type'] = 'private';
362
363			if ($mapping['propertyType'] == 'public') {
364			$response['property_type'] = 'public';
365			} elseif (isset($mapping['methods']['setter']) && isset($mapping['methods']['getter'])) {
366			$response['setter'] = $mapping['methods']['setter'];
367			$response['getter'] = $mapping['methods']['getter'];
368			}
369			} else {
370			return false;
371			}
372			return $response;
373			}
374
375			/**
376			* Adds ids to documents
377			*
378			* @param array $bulkQueries
379			*
380			* @param array $bulkResponse
381			*/
382			public function addIds(array $bulkQueries, $bulkResponse = [])
383			{
384			if (empty($bulkResponse)) {
385			$this->persistedObjects = [];
386			return;
387			}
388			$indexing = [];
389			foreach ($bulkQueries['body'] as $number => $query) {
390			if (isset($query['index']) && !isset($query['index']['_id'])) {
391			$indexing[] = $number / 2;
392			}
393			}
394			if (isset($bulkQueries['body'][0]['index'])) {
395			if (isset($this->persistedObjects)) {
396			$i = 0;
397			foreach ($this->persistedObjects as $document) {
398			if ($id_field_info = $this->getIdFieldInfo($document)) {
399			if ($id_field_info['property_type'] == 'public') {
400			$id_field = $id_field_info['id_property'];
401			$document->$id_field = $bulkResponse['items'][$indexing[$i]]['create']['_id'];
402			} elseif ($id_field_info['setter']) {
403			$method = $id_field_info['setter'];
404			$document->$method($bulkResponse['items'][$indexing[$i]]['create']['_id']);
405			}
406			}
407			$i++;
408			}
409			}
410			$this->persistedObjects = [];
411			}
412			}
413
414			/**
415			* Inserts the current query container to the index, used for bulk queries execution.
416			*
417			* @param array $params Parameters that will be passed to the flush or refresh queries.
418			*
419			* @return null\|array
420			*/
421			public function commit(array $params = [])
422			{
423			$this->isReadOnly('Commit');
424
425			if (!empty($this->bulkQueries)) {
426			$bulkQueries = array_merge($this->bulkQueries, $this->bulkParams);
427			$this->bulkQueries = [];
428
429			$bulkResponse = $this->client->bulk($bulkQueries);
430
431			switch ($this->getCommitMode()) {
432			case 'flush':
433			$this->flush($params);
434			break;
435			case 'refresh':
436			$this->refresh($params);
437			break;
438			}
439
440			$this->addIds($bulkQueries, $bulkResponse);
			0 ignored issues – show Documentation introduced 2016-03-15 20:41 UTC by Report Bug Copy Issue Report `$bulkResponse` is of type `callable`, but the function expects a `array`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
441
442			return $bulkResponse;
443			}
444			return null;
445			}
446
447			/**
448			* Adds query to bulk queries container.
449			*
450			* @param string $operation One of: index, update, delete, create.
451			* @param string\|array $type Elasticsearch type name.
452			* @param array $query DSL to execute.
453			*
454			* @throws \InvalidArgumentException
455			*/
456			public function bulk($operation, $type, array $query)
457			{
458			$this->isReadOnly('Bulk');
459
460			if (!in_array($operation, ['index', 'create', 'update', 'delete'])) {
461			throw new \InvalidArgumentException('Wrong bulk operation selected');
462			}
463
464			$this->bulkQueries['body'][] = [
465			$operation => array_filter(
466			[
467			'_index' => $this->getIndexName(),
468			'_type' => $type,
469			'_id' => isset($query['_id']) ? $query['_id'] : null,
470			'_ttl' => isset($query['_ttl']) ? $query['_ttl'] : null,
471			'_parent' => isset($query['_parent']) ? $query['_parent'] : null,
472			]
473			),
474			];
475			unset($query['_id'], $query['_ttl'], $query['_parent']);
476
477			switch ($operation) {
478			case 'index':
479			case 'create':
480			case 'update':
481			$this->bulkQueries['body'][] = $query;
482			break;
483			case 'delete':
484			// Body for delete operation is not needed to apply.
485			default:
486			// Do nothing.
487			break;
488			}
489
490			// We are using counter because there is to difficult to resolve this from bulkQueries array.
491			$this->bulkCount++;
492
493			if ($this->bulkCommitSize === $this->bulkCount) {
494			$this->commit();
495			$this->bulkCount = 0;
496			}
497			}
498
499			/**
500			* Optional setter to change bulk query params.
501			*
502			* @param array $params Possible keys:
503			* ['consistency'] = (enum) Explicit write consistency setting for the operation.
504			* ['refresh'] = (boolean) Refresh the index after performing the operation.
505			* ['replication'] = (enum) Explicitly set the replication type.
506			*/
507			public function setBulkParams(array $params)
508			{
509			$this->bulkParams = $params;
510			}
511
512			/**
513			* Creates fresh elasticsearch index.
514			*
515			* @param bool $noMapping Determines if mapping should be included.
516			*
517			* @return array
518			*/
519			public function createIndex($noMapping = false)
520			{
521			$this->isReadOnly('Create index');
522
523			if ($noMapping) {
524			unset($this->indexSettings['body']['mappings']);
525			}
526
527			return $this->getClient()->indices()->create($this->indexSettings);
528			}
529
530			/**
531			* Drops elasticsearch index.
532			*/
533			public function dropIndex()
534			{
535			$this->isReadOnly('Drop index');
536
537			return $this->getClient()->indices()->delete(['index' => $this->getIndexName()]);
538			}
539
540			/**
541			* Tries to drop and create fresh elasticsearch index.
542			*
543			* @param bool $noMapping Determines if mapping should be included.
544			*
545			* @return array
546			*/
547			public function dropAndCreateIndex($noMapping = false)
548			{
549			try {
550			$this->dropIndex();
551			} catch (\Exception $e) {
552			// Do nothing, our target is to create new index.
553			}
554
555			return $this->createIndex($noMapping);
556			}
557
558			/**
559			* Checks if connection index is already created.
560			*
561			* @return bool
562			*/
563			public function indexExists()
564			{
565			return $this->getClient()->indices()->exists(['index' => $this->getIndexName()]);
566			}
567
568			/**
569			* Returns index name this connection is attached to.
570			*
571			* @return string
572			*/
573			public function getIndexName()
574			{
575			return $this->indexSettings['index'];
576			}
577
578			/**
579			* Sets index name for this connection.
580			*
581			* @param string $name
582			*/
583			public function setIndexName($name)
584			{
585			$this->indexSettings['index'] = $name;
586			}
587
588			/**
589			* Returns Elasticsearch version number.
590			*
591			* @return string
592			*/
593			public function getVersionNumber()
594			{
595			return $this->client->info()['version']['number'];
596			}
597
598			/**
599			* Clears elasticsearch client cache.
600			*/
601			public function clearCache()
602			{
603			$this->isReadOnly('Clear cache');
604
605			$this->getClient()->indices()->clearCache(['index' => $this->getIndexName()]);
606			}
607
608			/**
609			* Set connection to read only state.
610			*
611			* @param bool $readOnly
612			*/
613			public function setReadOnly($readOnly)
614			{
615			$this->readOnly = $readOnly;
616			}
617
618			/**
619			* Checks if connection is read only.
620			*
621			* @param string $message Error message.
622			*
623			* @throws Forbidden403Exception
624			*/
625			public function isReadOnly($message = '')
626			{
627			if ($this->readOnly) {
628			throw new Forbidden403Exception("Manager is readonly! {$message} operation is not permitted.");
629			}
630			}
631
632			/**
633			* Returns a single document by ID. Returns NULL if document was not found.
634			*
635			* @param string $className Document class name or Elasticsearch type name
636			* @param string $id Document ID to find
637			*
638			* @return object
639			*/
640			public function find($className, $id)
641			{
642			$type = $this->resolveTypeName($className);
643
644			$params = [
645			'index' => $this->getIndexName(),
646			'type' => $type,
647			'id' => $id,
648			];
649
650			try {
651			$result = $this->getClient()->get($params);
652			} catch (Missing404Exception $e) {
653			return null;
654			}
655
656			return $this->getConverter()->convertToDocument($result, $this);
			0 ignored issues – show Documentation introduced 2015-12-29 10:45 UTC by Report Bug Copy Issue Report `$result` is of type `callable`, but the function expects a `array`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
657			}
658
659			/**
660			* Executes given search.
661			*
662			* @param array $types
663			* @param Search $search
664			* @param string $resultsType
665			*
666			* @return DocumentIterator\|RawIterator\|array
667			*/
668			public function execute($types, Search $search, $resultsType = Result::RESULTS_OBJECT)
669			{
670			foreach ($types as &$type) {
671			$type = $this->resolveTypeName($type);
672			}
673
674			$results = $this->search($types, $search->toArray(), $search->getQueryParams());
675
676			return $this->parseResult($results, $resultsType, $search->getScroll());
			0 ignored issues – show Documentation introduced 2015-12-29 10:45 UTC by Report Bug Copy Issue Report `$results` is of type `callable`, but the function expects a `array`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
677			}
678
679			/**
680			* Parses raw result.
681			*
682			* @param array $raw
683			* @param string $resultsType
684			* @param string $scrollDuration
685			*
686			* @return DocumentIterator\|RawIterator\|array
687			*
688			* @throws \Exception
689			*/
690			private function parseResult($raw, $resultsType, $scrollDuration = null)
691			{
692			$scrollConfig = [];
693			if (isset($raw['_scroll_id'])) {
694			$scrollConfig['_scroll_id'] = $raw['_scroll_id'];
695			$scrollConfig['duration'] = $scrollDuration;
696			}
697
698			switch ($resultsType) {
699			case Result::RESULTS_OBJECT:
700			return new DocumentIterator($raw, $this, $scrollConfig);
701			case Result::RESULTS_ARRAY:
702			return $this->convertToNormalizedArray($raw);
703			case Result::RESULTS_RAW:
704			return $raw;
705			case Result::RESULTS_RAW_ITERATOR:
706			return new RawIterator($raw, $this, $scrollConfig);
707			default:
708			throw new \Exception('Wrong results type selected');
709			}
710			}
711
712			/**
713			* Normalizes response array.
714			*
715			* @param array $data
716			*
717			* @return array
718			*/
719			private function convertToNormalizedArray($data)
720			{
721			if (array_key_exists('_source', $data)) {
722			return $data['_source'];
723			}
724
725			$output = [];
726
727			if (isset($data['hits']['hits'][0]['_source'])) {
728			foreach ($data['hits']['hits'] as $item) {
729			$output[] = $item['_source'];
730			}
731			} elseif (isset($data['hits']['hits'][0]['fields'])) {
732			foreach ($data['hits']['hits'] as $item) {
733			$output[] = array_map('reset', $item['fields']);
734			}
735			}
736
737			return $output;
738			}
739
740			/**
741			* Fetches next set of results.
742			*
743			* @param string $scrollId
744			* @param string $scrollDuration
745			* @param string $resultsType
746			*
747			* @return AbstractResultsIterator
748			*
749			* @throws \Exception
750			*/
751			public function scroll(
752			$scrollId,
753			$scrollDuration = '5m',
754			$resultsType = Result::RESULTS_OBJECT
755			) {
756			$results = $this->getClient()->scroll(['scroll_id' => $scrollId, 'scroll' => $scrollDuration]);
757
758			return $this->parseResult($results, $resultsType, $scrollDuration);
			0 ignored issues – show Documentation introduced 2015-12-29 10:45 UTC by Report Bug Copy Issue Report `$results` is of type `callable`, but the function expects a `array`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
759			}
760
761			/**
762			* Clears scroll.
763			*
764			* @param string $scrollId
765			*/
766			public function clearScroll($scrollId)
767			{
768			$this->getClient()->clearScroll(['scroll_id' => $scrollId]);
769			}
770
771			/**
772			* Resolves type name by class name.
773			*
774			* @param string $className
775			*
776			* @return string
777			*/
778			private function resolveTypeName($className)
779			{
780			if (strpos($className, ':') !== false \|\| strpos($className, '\\') !== false) {
781			return $this->getMetadataCollector()->getDocumentType($className);
782			}
783
784			return $className;
785			}
786			}
787

ongr-io / ElasticsearchBundle

Pull Request — master (#589)

Manager::persist() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like