Manager::updateIndex() - Code Metrics - Inspection of "Update elasticsearch index if exists" - ongr-io/ElasticsearchBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#780)

unknown

created 2017-04-10 13:33 UTC

Manager::updateIndex() A

↳ Parent: Manager

Complexity

Conditions	2
Paths	2

Size

Total Lines	14
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	14
rs	9.4285
c	0
b	0
f	0
cc	2
eloc	8
nc	2
nop	0

<?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\Missing404Exception;
use ONGR\ElasticsearchBundle\Event\Events;
use ONGR\ElasticsearchBundle\Event\BulkEvent;
use ONGR\ElasticsearchBundle\Event\CommitEvent;
use ONGR\ElasticsearchBundle\Exception\BulkWithErrorsException;
use ONGR\ElasticsearchBundle\Mapping\MetadataCollector;
use ONGR\ElasticsearchBundle\Result\Converter;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
use Symfony\Component\Stopwatch\Stopwatch;

/**
 * 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 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';

    /**
     * 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;

    /**
     * @var EventDispatcherInterface
     */
    private $eventDispatcher;

    /**
     * @var Stopwatch
     */
    private $stopwatch;

    /**
     * @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;
    }

    /**
     * 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;
    }

    /**
     * @param EventDispatcherInterface $eventDispatcher
     */
    public function setEventDispatcher(EventDispatcherInterface $eventDispatcher)
    {
        $this->eventDispatcher = $eventDispatcher;
    }

    /**
     * @param Stopwatch $stopwatch
     */
    public function setStopwatch(Stopwatch $stopwatch)
    {
        $this->stopwatch = $stopwatch;
    }

    /**
     * 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.');
        }

        $directory = null;

        if (strpos($className, ':')) {
            $bundle = explode(':', $className)[0];

            if (isset($this->config['mappings'][$bundle]['document_dir'])) {
                $directory = $this->config['mappings'][$bundle]['document_dir'];
            }
        }

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

        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();
        
        if (!empty($types)) {
            $params['type'] = implode(',', $types);
        }
        
        $params['body'] = $query;

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

        $this->stopwatch('start', 'search');
        $result = $this->client->search($params);
        $this->stopwatch('stop', 'search');

        return $result;
    }

    /**
     * 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);
    }

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

        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, $data);
    }

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

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

    /**
     * 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
     *
     * @throws BulkWithErrorsException
     */
    public function commit(array $params = [])
    {
        if (!empty($this->bulkQueries)) {
            $bulkQueries = array_merge($this->bulkQueries, $this->bulkParams);
            $bulkQueries['index']['_index'] = $this->getIndexName();
            $this->eventDispatcher->dispatch(
                Events::PRE_COMMIT,
                new CommitEvent($this->getCommitMode(), $bulkQueries)
            );

            $this->stopwatch('start', 'bulk');
            $bulkResponse = $this->client->bulk($bulkQueries);
            $this->stopwatch('stop', 'bulk');

            if ($bulkResponse['errors']) {
                throw new BulkWithErrorsException(
                    json_encode($bulkResponse),
                    0,
                    null,
                    $bulkResponse
function acceptsInteger($int) { }

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

// Instead of
acceptsInteger($x);

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

            $this->bulkQueries = [];
            $this->bulkCount = 0;

            $this->stopwatch('start', 'refresh');

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

            $this->eventDispatcher->dispatch(
                Events::POST_COMMIT,
                new CommitEvent($this->getCommitMode(), $bulkResponse)
function acceptsInteger($int) { }

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

// Instead of
acceptsInteger($x);

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

            $this->stopwatch('stop', 'refresh');

            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
     *
     * @return null|array
     */
    public function bulk($operation, $type, array $query)
    {
        if (!in_array($operation, ['index', 'create', 'update', 'delete'])) {
            throw new \InvalidArgumentException('Wrong bulk operation selected');
        }

        $this->eventDispatcher->dispatch(
            Events::BULK,
            new BulkEvent($operation, $type, $query)
        );

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

        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++;

        $response = null;

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

        return $response;
    }

    /**
     * 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)
    {
        if ($noMapping) {
            unset($this->indexSettings['body']['mappings']);
        }

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

    /**
     * Update elasticsearch index if exists.
     *
     * @return array
     */
    public function updateIndex()
    {
        $params = [];

        foreach ($this->getIndexMappings() as $key => $mapping) {
            $params = [
                'index' => $this->getIndexName(),
                'type' => $key,
                'body' => $mapping
            ];
        }

        return $this->getClient()->indices()->putMapping($params);
    }

    /**
     * Drops elasticsearch index.
     */
    public function dropIndex()
    {
        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 mappings of the index for this connection.
     *
     * @return array
     */
    public function getIndexMappings()
    {
        return $this->indexSettings['body']['mappings'];
    }

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

    /**
     * Clears elasticsearch client cache.
     */
    public function clearCache()
    {
        $this->getClient()->indices()->clearCache(['index' => $this->getIndexName()]);
    }

    /**
     * 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
     * @param string $routing   Custom routing for the document
     *
     * @return object
     */
    public function find($className, $id, $routing = null)
    {
        $type = $this->resolveTypeName($className);

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

        if ($routing) {
''   == false // true
''   == null  // true
'ab' == false // false
'ab' == null  // false

// It is often better to use strict comparison
'' === false // false
'' === null  // false
            $params['routing'] = $routing;
        }

        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);
    }

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

        return $results;
    }

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

    /**
     * Calls "Get Settings API" in Elasticsearch and will return you the currently configured settings.
     *
     * return array
     */
    public function getSettings()
    {
        return $this->getClient()->indices()->getSettings(['index' => $this->getIndexName()]);
    }

    /**
     * Gets Elasticsearch aliases information.
     * @param $params
     *
     * @return array
     */
    public function getAliases($params = [])
    {
        return $this->getClient()->indices()->getAliases(array_merge(['index' => $this->getIndexName()], $params));
    }

    /**
     * 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;
    }

    /**
     * Starts and stops an event in the stopwatch
     *
     * @param string $action   only 'start' and 'stop'
     * @param string $name     name of the event
     */
    private function stopwatch($action, $name)
    {
        if (isset($this->stopwatch)) {
            $this->stopwatch->$action('ongr_es: '.$name, 'ongr_es');
        }
    }
}


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\Missing404Exception;
16			use ONGR\ElasticsearchBundle\Event\Events;
17			use ONGR\ElasticsearchBundle\Event\BulkEvent;
18			use ONGR\ElasticsearchBundle\Event\CommitEvent;
19			use ONGR\ElasticsearchBundle\Exception\BulkWithErrorsException;
20			use ONGR\ElasticsearchBundle\Mapping\MetadataCollector;
21			use ONGR\ElasticsearchBundle\Result\Converter;
22			use Symfony\Component\EventDispatcher\EventDispatcherInterface;
23			use Symfony\Component\Stopwatch\Stopwatch;
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 array Container for bulk queries
52			*/
53			private $bulkQueries = [];
54
55			/**
56			* @var array Holder for consistency, refresh and replication parameters
57			*/
58			private $bulkParams = [];
59
60			/**
61			* @var array
62			*/
63			private $indexSettings;
64
65			/**
66			* @var MetadataCollector
67			*/
68			private $metadataCollector;
69
70			/**
71			* After commit to make data available the refresh or flush operation is needed
72			* so one of those methods has to be defined, the default is refresh.
73			*
74			* @var string
75			*/
76			private $commitMode = 'refresh';
77
78			/**
79			* The size that defines after how much document inserts call commit function.
80			*
81			* @var int
82			*/
83			private $bulkCommitSize = 100;
84
85			/**
86			* Container to count how many documents was passed to the bulk query.
87			*
88			* @var int
89			*/
90			private $bulkCount = 0;
91
92			/**
93			* @var Repository[] Repository local cache
94			*/
95			private $repositories;
96
97			/**
98			* @var EventDispatcherInterface
99			*/
100			private $eventDispatcher;
101
102			/**
103			* @var Stopwatch
104			*/
105			private $stopwatch;
106
107			/**
108			* @param string $name Manager name
109			* @param array $config Manager configuration
110			* @param Client $client
111			* @param array $indexSettings
112			* @param MetadataCollector $metadataCollector
113			* @param Converter $converter
114			*/
115			public function __construct(
116			$name,
117			array $config,
118			$client,
119			array $indexSettings,
120			$metadataCollector,
121			$converter
122			) {
123			$this->name = $name;
124			$this->config = $config;
125			$this->client = $client;
126			$this->indexSettings = $indexSettings;
127			$this->metadataCollector = $metadataCollector;
128			$this->converter = $converter;
129			}
130
131			/**
132			* Returns Elasticsearch connection.
133			*
134			* @return Client
135			*/
136			public function getClient()
137			{
138			return $this->client;
139			}
140
141			/**
142			* @return string
143			*/
144			public function getName()
145			{
146			return $this->name;
147			}
148
149			/**
150			* @return array
151			*/
152			public function getConfig()
153			{
154			return $this->config;
155			}
156
157			/**
158			* @param EventDispatcherInterface $eventDispatcher
159			*/
160			public function setEventDispatcher(EventDispatcherInterface $eventDispatcher)
161			{
162			$this->eventDispatcher = $eventDispatcher;
163			}
164
165			/**
166			* @param Stopwatch $stopwatch
167			*/
168			public function setStopwatch(Stopwatch $stopwatch)
169			{
170			$this->stopwatch = $stopwatch;
171			}
172
173			/**
174			* Returns repository by document class.
175			*
176			* @param string $className FQCN or string in Bundle:Document format
177			*
178			* @return Repository
179			*/
180			public function getRepository($className)
181			{
182			if (!is_string($className)) {
183			throw new \InvalidArgumentException('Document class must be a string.');
184			}
185
186			$directory = null;
187
188			if (strpos($className, ':')) {
189			$bundle = explode(':', $className)[0];
190
191			if (isset($this->config['mappings'][$bundle]['document_dir'])) {
192			$directory = $this->config['mappings'][$bundle]['document_dir'];
193			}
194			}
195
196			$namespace = $this->getMetadataCollector()->getClassName($className, $directory);
197
198			if (isset($this->repositories[$namespace])) {
199			return $this->repositories[$namespace];
200			}
201
202			$repository = $this->createRepository($namespace);
203			$this->repositories[$namespace] = $repository;
204
205			return $repository;
206			}
207
208			/**
209			* @return MetadataCollector
210			*/
211			public function getMetadataCollector()
212			{
213			return $this->metadataCollector;
214			}
215
216			/**
217			* @return Converter
218			*/
219			public function getConverter()
220			{
221			return $this->converter;
222			}
223
224			/**
225			* @return string
226			*/
227			public function getCommitMode()
228			{
229			return $this->commitMode;
230			}
231
232			/**
233			* @param string $commitMode
234			*/
235			public function setCommitMode($commitMode)
236			{
237			if ($commitMode === 'refresh' \|\| $commitMode === 'flush' \|\| $commitMode === 'none') {
238			$this->commitMode = $commitMode;
239			} else {
240			throw new \LogicException('The commit method must be either refresh, flush or none.');
241			}
242			}
243
244			/**
245			* @return int
246			*/
247			public function getBulkCommitSize()
248			{
249			return $this->bulkCommitSize;
250			}
251
252			/**
253			* @param int $bulkCommitSize
254			*/
255			public function setBulkCommitSize($bulkCommitSize)
256			{
257			$this->bulkCommitSize = $bulkCommitSize;
258			}
259
260			/**
261			* Creates a repository.
262			*
263			* @param string $className
264			*
265			* @return Repository
266			*/
267			private function createRepository($className)
268			{
269			return new Repository($this, $className);
270			}
271
272			/**
273			* Executes search query in the index.
274			*
275			* @param array $types List of types to search in.
276			* @param array $query Query to execute.
277			* @param array $queryStringParams Query parameters.
278			*
279			* @return array
280			*/
281			public function search(array $types, array $query, array $queryStringParams = [])
282			{
283			$params = [];
284			$params['index'] = $this->getIndexName();
285
286			if (!empty($types)) {
287			$params['type'] = implode(',', $types);
288			}
289
290			$params['body'] = $query;
291
292			if (!empty($queryStringParams)) {
293			$params = array_merge($queryStringParams, $params);
294			}
295
296			$this->stopwatch('start', 'search');
297			$result = $this->client->search($params);
298			$this->stopwatch('stop', 'search');
299
300			return $result;
301			}
302
303			/**
304			* Adds document to next flush.
305			*
306			* @param object $document
307			*/
308			public function persist($document)
309			{
310			$documentArray = $this->converter->convertToArray($document);
311			$type = $this->getMetadataCollector()->getDocumentType(get_class($document));
312
313			$this->bulk('index', $type, $documentArray);
314			}
315
316			/**
317			* Adds document for removal.
318			*
319			* @param object $document
320			*/
321			public function remove($document)
322			{
323			$data = $this->converter->convertToArray($document, [], ['_id', '_routing']);
324
325			if (!isset($data['_id'])) {
326			throw new \LogicException(
327			'In order to use remove() method document class must have property with @Id annotation.'
328			);
329			}
330
331			$type = $this->getMetadataCollector()->getDocumentType(get_class($document));
332
333			$this->bulk('delete', $type, $data);
334			}
335
336			/**
337			* Flushes elasticsearch index.
338			*
339			* @param array $params
340			*
341			* @return array
342			*/
343			public function flush(array $params = [])
344			{
345			return $this->client->indices()->flush(array_merge(['index' => $this->getIndexName()], $params));
346			}
347
348			/**
349			* Refreshes elasticsearch index.
350			*
351			* @param array $params
352			*
353			* @return array
354			*/
355			public function refresh(array $params = [])
356			{
357			return $this->client->indices()->refresh(array_merge(['index' => $this->getIndexName()], $params));
358			}
359
360			/**
361			* Inserts the current query container to the index, used for bulk queries execution.
362			*
363			* @param array $params Parameters that will be passed to the flush or refresh queries.
364			*
365			* @return null\|array
366			*
367			* @throws BulkWithErrorsException
368			*/
369			public function commit(array $params = [])
370			{
371			if (!empty($this->bulkQueries)) {
372			$bulkQueries = array_merge($this->bulkQueries, $this->bulkParams);
373			$bulkQueries['index']['_index'] = $this->getIndexName();
374			$this->eventDispatcher->dispatch(
375			Events::PRE_COMMIT,
376			new CommitEvent($this->getCommitMode(), $bulkQueries)
377			);
378
379			$this->stopwatch('start', 'bulk');
380			$bulkResponse = $this->client->bulk($bulkQueries);
381			$this->stopwatch('stop', 'bulk');
382
383			if ($bulkResponse['errors']) {
384			throw new BulkWithErrorsException(
385			json_encode($bulkResponse),
386			0,
387			null,
388			$bulkResponse
			0 ignored issues – show Documentation introduced 2016-09-01 08:37 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...
389			);
390			}
391
392			$this->bulkQueries = [];
393			$this->bulkCount = 0;
394
395			$this->stopwatch('start', 'refresh');
396
397			switch ($this->getCommitMode()) {
398			case 'flush':
399			$this->flush($params);
400			break;
401			case 'refresh':
402			$this->refresh($params);
403			break;
404			}
405
406			$this->eventDispatcher->dispatch(
407			Events::POST_COMMIT,
408			new CommitEvent($this->getCommitMode(), $bulkResponse)
			0 ignored issues – show Documentation introduced 2016-07-07 10:33 UTC by Report Bug Copy Issue Report `$bulkResponse` is of type `callable`, but the function expects a `array\|null`. 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...
409			);
410
411			$this->stopwatch('stop', 'refresh');
412
413			return $bulkResponse;
414			}
415
416			return null;
417			}
418
419			/**
420			* Adds query to bulk queries container.
421			*
422			* @param string $operation One of: index, update, delete, create.
423			* @param string\|array $type Elasticsearch type name.
424			* @param array $query DSL to execute.
425			*
426			* @throws \InvalidArgumentException
427			*
428			* @return null\|array
429			*/
430			public function bulk($operation, $type, array $query)
431			{
432			if (!in_array($operation, ['index', 'create', 'update', 'delete'])) {
433			throw new \InvalidArgumentException('Wrong bulk operation selected');
434			}
435
436			$this->eventDispatcher->dispatch(
437			Events::BULK,
438			new BulkEvent($operation, $type, $query)
439			);
440
441			$this->bulkQueries['body'][] = [
442			$operation => array_filter(
443			[
444			'_type' => $type,
445			'_id' => isset($query['_id']) ? $query['_id'] : null,
446			'_ttl' => isset($query['_ttl']) ? $query['_ttl'] : null,
447			'_routing' => isset($query['_routing']) ? $query['_routing'] : null,
448			'_parent' => isset($query['_parent']) ? $query['_parent'] : null,
449			]
450			),
451			];
452			unset($query['_id'], $query['_ttl'], $query['_parent'], $query['_routing']);
453
454			switch ($operation) {
455			case 'index':
456			case 'create':
457			case 'update':
458			$this->bulkQueries['body'][] = $query;
459			break;
460			case 'delete':
461			// Body for delete operation is not needed to apply.
462			default:
463			// Do nothing.
464			break;
465			}
466
467			// We are using counter because there is to difficult to resolve this from bulkQueries array.
468			$this->bulkCount++;
469
470			$response = null;
471
472			if ($this->bulkCommitSize === $this->bulkCount) {
473			$response = $this->commit();
474			}
475
476			return $response;
477			}
478
479			/**
480			* Optional setter to change bulk query params.
481			*
482			* @param array $params Possible keys:
483			* ['consistency'] = (enum) Explicit write consistency setting for the operation.
484			* ['refresh'] = (boolean) Refresh the index after performing the operation.
485			* ['replication'] = (enum) Explicitly set the replication type.
486			*/
487			public function setBulkParams(array $params)
488			{
489			$this->bulkParams = $params;
490			}
491
492			/**
493			* Creates fresh elasticsearch index.
494			*
495			* @param bool $noMapping Determines if mapping should be included.
496			*
497			* @return array
498			*/
499			public function createIndex($noMapping = false)
500			{
501			if ($noMapping) {
502			unset($this->indexSettings['body']['mappings']);
503			}
504
505			return $this->getClient()->indices()->create($this->indexSettings);
506			}
507
508			/**
509			* Update elasticsearch index if exists.
510			*
511			* @return array
512			*/
513			public function updateIndex()
514			{
515			$params = [];
516
517			foreach ($this->getIndexMappings() as $key => $mapping) {
518			$params = [
519			'index' => $this->getIndexName(),
520			'type' => $key,
521			'body' => $mapping
522			];
523			}
524
525			return $this->getClient()->indices()->putMapping($params);
526			}
527
528			/**
529			* Drops elasticsearch index.
530			*/
531			public function dropIndex()
532			{
533			return $this->getClient()->indices()->delete(['index' => $this->getIndexName()]);
534			}
535
536			/**
537			* Tries to drop and create fresh elasticsearch index.
538			*
539			* @param bool $noMapping Determines if mapping should be included.
540			*
541			* @return array
542			*/
543			public function dropAndCreateIndex($noMapping = false)
544			{
545			try {
546			$this->dropIndex();
547			} catch (\Exception $e) {
548			// Do nothing, our target is to create new index.
549			}
550
551			return $this->createIndex($noMapping);
552			}
553
554			/**
555			* Checks if connection index is already created.
556			*
557			* @return bool
558			*/
559			public function indexExists()
560			{
561			return $this->getClient()->indices()->exists(['index' => $this->getIndexName()]);
562			}
563
564			/**
565			* Returns index name this connection is attached to.
566			*
567			* @return string
568			*/
569			public function getIndexName()
570			{
571			return $this->indexSettings['index'];
572			}
573
574			/**
575			* Sets index name for this connection.
576			*
577			* @param string $name
578			*/
579			public function setIndexName($name)
580			{
581			$this->indexSettings['index'] = $name;
582			}
583
584			/**
585			* Returns mappings of the index for this connection.
586			*
587			* @return array
588			*/
589			public function getIndexMappings()
590			{
591			return $this->indexSettings['body']['mappings'];
592			}
593
594			/**
595			* Returns Elasticsearch version number.
596			*
597			* @return string
598			*/
599			public function getVersionNumber()
600			{
601			return $this->client->info()['version']['number'];
602			}
603
604			/**
605			* Clears elasticsearch client cache.
606			*/
607			public function clearCache()
608			{
609			$this->getClient()->indices()->clearCache(['index' => $this->getIndexName()]);
610			}
611
612			/**
613			* Returns a single document by ID. Returns NULL if document was not found.
614			*
615			* @param string $className Document class name or Elasticsearch type name
616			* @param string $id Document ID to find
617			* @param string $routing Custom routing for the document
618			*
619			* @return object
620			*/
621			public function find($className, $id, $routing = null)
622			{
623			$type = $this->resolveTypeName($className);
624
625			$params = [
626			'index' => $this->getIndexName(),
627			'type' => $type,
628			'id' => $id,
629			];
630
631			if ($routing) {
			0 ignored issues – show Bug Best Practice introduced 2016-07-13 06:48 UTC by Report Bug Copy Issue Report The expression `$routing` of type `string\|null` is loosely compared to `true`; this is ambiguous if the string can be empty. You might want to explicitly use `!== null` instead. In PHP, under loose comparison (like `==`, or `!=`, or `switch` conditions), values of different types might be equal. For `string` values, the empty string `''` is a special case, in particular the following results might be unexpected: '' == false // true '' == null // true 'ab' == false // false 'ab' == null // false // It is often better to use strict comparison '' === false // false '' === null // false Loading history...
632			$params['routing'] = $routing;
633			}
634
635			try {
636			$result = $this->getClient()->get($params);
637			} catch (Missing404Exception $e) {
638			return null;
639			}
640
641			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...
642			}
643
644			/**
645			* Fetches next set of results.
646			*
647			* @param string $scrollId
648			* @param string $scrollDuration
649			*
650			* @return mixed
651			*
652			* @throws \Exception
653			*/
654			public function scroll(
655			$scrollId,
656			$scrollDuration = '5m'
657			) {
658			$results = $this->getClient()->scroll(['scroll_id' => $scrollId, 'scroll' => $scrollDuration]);
659
660			return $results;
661			}
662
663			/**
664			* Clears scroll.
665			*
666			* @param string $scrollId
667			*/
668			public function clearScroll($scrollId)
669			{
670			$this->getClient()->clearScroll(['scroll_id' => $scrollId]);
671			}
672
673			/**
674			* Calls "Get Settings API" in Elasticsearch and will return you the currently configured settings.
675			*
676			* return array
677			*/
678			public function getSettings()
679			{
680			return $this->getClient()->indices()->getSettings(['index' => $this->getIndexName()]);
681			}
682
683			/**
684			* Gets Elasticsearch aliases information.
685			* @param $params
686			*
687			* @return array
688			*/
689			public function getAliases($params = [])
690			{
691			return $this->getClient()->indices()->getAliases(array_merge(['index' => $this->getIndexName()], $params));
692			}
693
694			/**
695			* Resolves type name by class name.
696			*
697			* @param string $className
698			*
699			* @return string
700			*/
701			private function resolveTypeName($className)
702			{
703			if (strpos($className, ':') !== false \|\| strpos($className, '\\') !== false) {
704			return $this->getMetadataCollector()->getDocumentType($className);
705			}
706
707			return $className;
708			}
709
710			/**
711			* Starts and stops an event in the stopwatch
712			*
713			* @param string $action only 'start' and 'stop'
714			* @param string $name name of the event
715			*/
716			private function stopwatch($action, $name)
717			{
718			if (isset($this->stopwatch)) {
719			$this->stopwatch->$action('ongr_es: '.$name, 'ongr_es');
720			}
721			}
722			}
723

ongr-io / ElasticsearchBundle

Pull Request — master (#780)

Manager::updateIndex() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like