Completed
Push — 1.0 ( f45751...770e00 )
by Simonas
12:41 queued 10:12
created

Manager::clearScroll()   A

Complexity

Conditions 1
Paths 1

Size

Total Lines 4
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Importance

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