FOS\ElasticaBundle\Persister\AsyncPagerPersister

Complexity

Total Complexity

7

Size/Duplication

Total Lines	79
Duplicated Lines	0 %

Coupling/Cohesion

Components	2
Dependencies	7

Test Coverage

Coverage

94.12%

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	9	1
A	insertPage()	0	24	3
A	insert()	0	20	3

<?php

/*
 * This file is part of the FOSElasticaBundle package.
 *
 * (c) FriendsOfSymfony <https://friendsofsymfony.github.com/>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace FOS\ElasticaBundle\Persister;

use FOS\ElasticaBundle\Message\AsyncPersistPage;
use FOS\ElasticaBundle\Provider\PagerInterface;
use FOS\ElasticaBundle\Provider\PagerProviderRegistry;
use Symfony\Component\Messenger\MessageBusInterface;

final class AsyncPagerPersister implements PagerPersisterInterface
{
    public const NAME = 'async';
    private const DEFAULT_PAGE_SIZE = 100;

    /**
     * @var PagerPersisterRegistry
     */
    private $pagerPersisterRegistry;

    /**
     * @var PagerProviderRegistry
     */
    private $pagerProviderRegistry;

    /**
     * @var MessageBusInterface
     */
    private $messageBus;

    public function __construct(
        PagerPersisterRegistry $pagerPersisterRegistry,
        PagerProviderRegistry $pagerProviderRegistry,
        MessageBusInterface $messageBus
    ) {
        $this->pagerPersisterRegistry = $pagerPersisterRegistry;
        $this->pagerProviderRegistry = $pagerProviderRegistry;
        $this->messageBus = $messageBus;
    }

    /**
     * {@inheritdoc}
     */
    public function insert(PagerInterface $pager, array $options = []): void
    {
        $pager->setMaxPerPage(empty($options['max_per_page']) ? self::DEFAULT_PAGE_SIZE : $options['max_per_page']);

        $options = \array_replace([
            'max_per_page' => $pager->getMaxPerPage(),
            'first_page' => $pager->getCurrentPage(),
            'last_page' => $pager->getNbPages(),
        ], $options);

        $pager->setCurrentPage($options['first_page']);

        $lastPage = \min($options['last_page'], $pager->getNbPages());
        $page = $pager->getCurrentPage();
        do {
            $this->messageBus->dispatch(new AsyncPersistPage($page, $options));

            ++$page;
        } while ($page <= $lastPage);
    }

    public function insertPage(int $page, array $options = []): int
    {
        if (!isset($options['indexName'])) {
            throw new \RuntimeException('Invalid call. $options is missing the indexName key.');
        }
        if (!isset($options['max_per_page'])) {
            throw new \RuntimeException('Invalid call. $options is missing the max_per_page key.');
        }

        $options['first_page'] = $page;
        $options['last_page'] = $page;

        $provider = $this->pagerProviderRegistry->getProvider($options['indexName']);
        $pager = $provider->provide($options);
        $pager->setMaxPerPage($options['max_per_page']);
        $pager->setCurrentPage($options['first_page']);

        $objectCount = \count($pager->getCurrentPageResults());

        $pagerPersister = $this->pagerPersisterRegistry->getPagerPersister(InPlacePagerPersister::NAME);
        $pagerPersister->insert($pager, $options);

        return $objectCount;
    }
}