Elastica\Bulk

Complexity

Total Complexity

52

Size/Duplication

Total Lines	368
Duplicated Lines	5.43 %

Coupling/Cohesion

Components	1
Dependencies	10

Importance

Changes

0

21 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	4	1
A	setIndex()	10	10	2
A	getIndex()	0	4	1
A	hasIndex()	0	4	2
A	getPath()	0	10	2
A	addAction()	0	6	1
A	addActions()	0	8	2
A	getActions()	0	4	1
A	addDocument()	0	6	1
A	addDocuments()	0	8	2
A	addScript()	0	6	1
A	addScripts()	0	8	2
A	addData()	0	18	5
B	addRawData()	0	31	7
A	setRequestParam()	0	6	1
A	setShardTimeout()	0	4	1
A	__toString()	0	4	1
A	toString()	0	9	2
A	toArray()	0	11	3
A	send()	0	9	1
C	_processResponse()	10	45	13

<?php

namespace Elastica;

use Elastica\Bulk\Action;
use Elastica\Bulk\Action\AbstractDocument as AbstractDocumentAction;
use Elastica\Bulk\Response as BulkResponse;
use Elastica\Bulk\ResponseSet;
use Elastica\Exception\Bulk\ResponseException;
use Elastica\Exception\Bulk\ResponseException as BulkResponseException;
use Elastica\Exception\InvalidException;
use Elastica\Script\AbstractScript;

class Bulk
{
    const DELIMITER = "\n";

    /**
     * @var Client
     */
    protected $_client;

    /**
     * @var Action[]
     */
    protected $_actions = [];

    /**
     * @var string|null
     */
    protected $_index;

    /**
     * @var string|null
     */
    protected $_type;

    /**
     * @var array request parameters to the bulk api
     */
    protected $_requestParams = [];

    /**
     * @param Client $client
     */
    public function __construct(Client $client)
    {
        $this->_client = $client;
    }

    /**
     * @param string|Index $index
     *
     * @return $this
     */
    public function setIndex($index): self

This method seems to be duplicated in your project.

    {
        if ($index instanceof Index) {
            $index = $index->getName();
        }

        $this->_index = (string) $index;

        return $this;
    }

    /**
     * @return string|null
     */
    public function getIndex()
    {
        return $this->_index;
    }

    /**
     * @return bool
     */
    public function hasIndex(): bool
    {
        return null !== $this->getIndex() && '' !== $this->getIndex();
    }

    /**
     * @return string
     */
    public function getPath(): string
    {
        $path = '';
        if ($this->hasIndex()) {
            $path .= $this->getIndex().'/';
        }
        $path .= '_bulk';

        return $path;
    }

    /**
     * @param Action $action
     *
     * @return $this
     */
    public function addAction(Action $action): self
    {
        $this->_actions[] = $action;

        return $this;
    }

    /**
     * @param Action[] $actions
     *
     * @return $this
     */
    public function addActions(array $actions): self
    {
        foreach ($actions as $action) {
            $this->addAction($action);
        }

        return $this;
    }

    /**
     * @return Action[]
     */
    public function getActions(): array
    {
        return $this->_actions;
    }

    /**
     * @param Document $document
     * @param string   $opType
     *
     * @return $this
     */
    public function addDocument(Document $document, string $opType = null): self
    {
        $action = AbstractDocumentAction::create($document, $opType);

        return $this->addAction($action);
    }

    /**
     * @param Document[] $documents
     * @param string     $opType
     *
     * @return $this
     */
    public function addDocuments(array $documents, string $opType = null): self
    {
        foreach ($documents as $document) {
            $this->addDocument($document, $opType);
        }

        return $this;
    }

    /**
     * @param AbstractScript $script
     * @param string         $opType
     *
     * @return $this
     */
    public function addScript(AbstractScript $script, string $opType = null): self
    {
        $action = AbstractDocumentAction::create($script, $opType);

        return $this->addAction($action);
    }

    /**
     * @param Document[] $scripts
     * @param string     $opType
     *
     * @return $this
     */
    public function addScripts(array $scripts, $opType = null): self
    {
        foreach ($scripts as $document) {
            $this->addScript($document, $opType);

$document is of type object<Elastica\Document>, but the function expects a object<Elastica\Script\AbstractScript>.

        }

        return $this;
    }

    /**
     * @param \Elastica\Script\AbstractScript|\Elastica\Document|array $data
     * @param string                                                   $opType
     *
     * @return $this
     */
    public function addData($data, string $opType = null)
    {
        if (!\is_array($data)) {
            $data = [$data];
        }

        foreach ($data as $actionData) {
            if ($actionData instanceof AbstractScript) {
                $this->addScript($actionData, $opType);
            } elseif ($actionData instanceof Document) {
                $this->addDocument($actionData, $opType);
            } else {
                throw new \InvalidArgumentException('Data should be a Document, a Script or an array containing Documents and/or Scripts');
            }
        }

        return $this;
    }

    /**
     * @param array $data
     *
     * @throws InvalidException
     *
     * @return $this
     */
    public function addRawData(array $data): self
    {
        foreach ($data as $row) {
            if (\is_array($row)) {
                $opType = \key($row);
                $metadata = \reset($row);
                if (Action::isValidOpType($opType)) {
                    // add previous action
                    if (isset($action)) {
                        $this->addAction($action);
                    }
                    $action = new Action($opType, $metadata);
                } elseif (isset($action)) {
                    $action->setSource($row);
                    $this->addAction($action);
                    $action = null;
                } else {
                    throw new InvalidException('Invalid bulk data, source must follow action metadata');
                }
            } else {
                throw new InvalidException('Invalid bulk data, should be array of array, Document or Bulk/Action');
            }
        }

        // add last action if available
        if (isset($action)) {
            $this->addAction($action);
        }

        return $this;
    }

    /**
     * Set a url parameter on the request bulk request.
     *
     * @param string $name  name of the parameter
     * @param mixed  $value value of the parameter
     *
     * @return $this
     */
    public function setRequestParam(string $name, $value): self
    {
        $this->_requestParams[$name] = $value;

        return $this;
    }

    /**
     * Set the amount of time that the request will wait the shards to come on line.
     * Requires Elasticsearch version >= 0.90.8.
     *
     * @param string $time timeout in Elasticsearch time format
     *
     * @return $this
     */
    public function setShardTimeout(string $time): self
    {
        return $this->setRequestParam('timeout', $time);
    }

    /**
     * @return string
     */
    public function __toString(): string
    {
        return $this->toString();
    }

    /**
     * @return string
     */
    public function toString(): string
    {
        $data = '';
        foreach ($this->getActions() as $action) {
            $data .= $action->toString();
        }

        return $data;
    }

    /**
     * @return array
     */
    public function toArray(): array
    {
        $data = [];
        foreach ($this->getActions() as $action) {
            foreach ($action->toArray() as $row) {
                $data[] = $row;
            }
        }

        return $data;
    }

    /**
     * @return ResponseSet
     */
    public function send(): ResponseSet
    {
        $path = $this->getPath();
        $data = $this->toString();

        $response = $this->_client->request($path, Request::POST, $data, $this->_requestParams, Request::NDJSON_CONTENT_TYPE);

        return $this->_processResponse($response);
    }

    /**
     * @param Response $response
     *
     * @throws ResponseException
     * @throws InvalidException
     *
     * @return ResponseSet
     */
    protected function _processResponse(Response $response): ResponseSet
    {
        $responseData = $response->getData();

        $actions = $this->getActions();

        $bulkResponses = [];

        if (isset($responseData['items']) && \is_array($responseData['items'])) {
            foreach ($responseData['items'] as $key => $item) {
                if (!isset($actions[$key])) {
                    throw new InvalidException('No response found for action #'.$key);
                }

                $action = $actions[$key];

                $opType = \key($item);
                $bulkResponseData = \reset($item);

                if ($action instanceof AbstractDocumentAction) {
                    $data = $action->getData();
                    if ($data instanceof Document && $data->isAutoPopulate()

This code seems to be duplicated across your project.

                        || $this->_client->getConfigValue(['document', 'autoPopulate'], false)
                    ) {
                        if (!$data->hasId() && isset($bulkResponseData['_id'])) {
                            $data->setId($bulkResponseData['_id']);
                        }
                        if (isset($bulkResponseData['_version'])) {
                            $data->setVersion($bulkResponseData['_version']);
                        }
                    }
                }

                $bulkResponses[] = new BulkResponse($bulkResponseData, $action, $opType);
            }
        }

        $bulkResponseSet = new ResponseSet($response, $bulkResponses);

        if ($bulkResponseSet->hasError()) {
            throw new BulkResponseException($bulkResponseSet);
        }

        return $bulkResponseSet;
    }
}