Manager::setEventDispatcher() - Code Metrics - Inspection of "Merge pull request #669 from einorler/issue661" - ongr-io/ElasticsearchBundle - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 51b986...23baca )

by Simonas

created 2016-08-22 07:15 UTC

Manager::setEventDispatcher() A

↳ Parent: Manager

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
c	1
b	0
f	1
dl	0
loc	4
rs	10
cc	1
eloc	2
nc	1
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\Missing404Exception;
use ONGR\ElasticsearchBundle\Event\Events;
use ONGR\ElasticsearchBundle\Event\BulkEvent;
use ONGR\ElasticsearchBundle\Event\PersistEvent;
use ONGR\ElasticsearchBundle\Event\CommitEvent;
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;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

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

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

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

    /**
     * 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(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
     */
    public function commit(array $params = [])
    {
        if (!empty($this->bulkQueries)) {
            $bulkQueries = array_merge($this->bulkQueries, $this->bulkParams);

            $this->eventDispatcher->dispatch(
                Events::PRE_COMMIT,
                new CommitEvent($this->getCommitMode(), $bulkQueries)
            );

            $bulkResponse = $this->client->bulk($bulkQueries);
            $this->bulkQueries = [];
            $this->bulkCount = 0;

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

            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(
                [
                    '_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++;

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

ongr-io / ElasticsearchBundle

Push — master ( 51b986...23baca )

Manager::setEventDispatcher() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like