Manager::dispatch() - Code Metrics - Inspection of "Add Symfony 5 compatibility" - ongr-io/ElasticsearchBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — 5.2 (#949)

by Alexander

created 2020-06-10 08:58 UTC

Manager::dispatch() A

↳ Parent: Manager

Complexity

Conditions	2
Paths	2

Size

Total Lines

Duplication

Lines	8
Ratio	100 %

Importance

Changes

Metric	Value
dl	8
loc	8
rs	10
c	0
b	0
f	0
cc	2
nc	2
nop	2

<?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\Event\PrePersistEvent;
use ONGR\ElasticsearchBundle\Exception\BulkWithErrorsException;
use ONGR\ElasticsearchBundle\Mapping\MetadataCollector;
use ONGR\ElasticsearchBundle\Result\Converter;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
use Symfony\Component\EventDispatcher\LegacyEventDispatcherProxy;
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();

        $resolvedTypes = [];
        foreach ($types as $type) {
            $resolvedTypes[] = $this->resolveTypeName($type);
        }

        if (!empty($resolvedTypes)) {
            $params['type'] = implode(',', $resolvedTypes);
        }

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

    /**
     * Execute search queries using multisearch api
     * $body - is array of requests described in elastic Multi Search API
     *
     * @param $body
     * @return array
     */
    public function msearch(array $body)
    {
        $result = $this->client->msearch(
            [
                'index' => $this->getIndexName(), // set default index
                'body' => $body
            ]
        );
        return $result;
    }

    /**
     * Adds document to next flush.
     *
     * @param object $document
     */
    public function persist($document)
    {
        $this->dispatch(
            Events::PRE_PERSIST,
            new PrePersistEvent($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->dispatch(
                Events::PRE_COMMIT,
                new CommitEvent($this->getCommitMode(), $bulkQueries)
            );

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

            if (isset($bulkResponse['errors']) && $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->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->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);
    }

    /**
     * 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 {
            if ($this->indexExists()) {
                $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');
        }
    }

    private function dispatch($eventName, $event)

    {
        if (class_exists(LegacyEventDispatcherProxy::class)) {
            return $this->eventDispatcher->dispatch($event, $eventName);
        } else {
            return $this->eventDispatcher->dispatch($eventName, $event);
        }
    }
}


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

ongr-io / ElasticsearchBundle

Pull Request — 5.2 (#949)

Manager::dispatch() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like