ClickHouseDB\Transport\Http

Complexity

Total Complexity

60

Size/Duplication

Total Lines	553
Duplicated Lines	0 %

Test Coverage

Coverage

92.35%

Importance

Changes

0

28 Methods

Rating	Name	Duplication	Size	Complexity
A	getRequestWrite()	0	4	1
B	makeRequest()	0	30	4
A	getConnectTimeOut()	0	3	1
A	verbose()	0	4	1
A	addQueryDegeneration()	0	4	1
A	executeAsync()	0	3	1
A	getUri()	0	6	2
A	getCurler()	0	3	1
A	setConnectTimeOut()	0	3	1
C	__findXClickHouseProgress()	0	25	8
A	cleanQueryDegeneration()	0	4	1
A	setHost()	0	7	2
A	__construct()	0	9	1
A	writeStreamData()	0	18	1
A	newRequest()	0	20	3
A	prepareWrite()	0	8	2
B	getUrl()	0	22	4
A	settings()	0	3	1
A	getCountPendingQueue()	0	3	1
A	prepareQuery()	0	9	2
A	setCurler()	0	3	1
A	prepareSelect()	0	10	2
B	writeAsyncCSV()	0	26	1
C	getRequestRead()	0	56	11
A	selectAsync()	0	5	1
A	setProgressFunction()	0	3	1
A	select()	0	5	1
A	write()	0	11	3

<?php

namespace ClickHouseDB\Transport;

use ClickHouseDB\Query\Degeneration;
use ClickHouseDB\Query\Query;
use ClickHouseDB\Query\WhereInFile;
use ClickHouseDB\Query\WriteToFile;
use ClickHouseDB\Settings;
use ClickHouseDB\Statement;

class Http
{
    /**
     * @var string
     */
    private $_username = null;

    /**
     * @var string
     */
    private $_password = null;

    /**
     * @var string
     */
    private $_host = '';

    /**
     * @var int
     */
    private $_port = 0;

    /**
     * @var bool
     */
    private $_verbose = false;

    /**
     * @var CurlerRolling
     */
    private $_curler = false;

    /**
     * @var Settings
     */
    private $_settings = false;

    /**
     * @var array
     */
    private $_query_degenerations = [];

    /**
     * Count seconds (int)
     *
     * @var int
     */
    private $_connectTimeOut = 5;

    /**
     * @var bool
     */
    private $xClickHouseProgress=false;

    /**
     * Http constructor.
     * @param $host
     * @param $port
     * @param $username
     * @param $password
     */
    public function __construct($host, $port, $username, $password)
    {
        $this->setHost($host, $port);

        $this->_username = $username;
        $this->_password = $password;
        $this->_settings = new Settings($this);

        $this->setCurler();
    }


    public function setCurler()
    {
        $this->_curler = new CurlerRolling();
    }

    /**
     * @return CurlerRolling
     */
    public function getCurler()
    {
        return $this->_curler;
    }

    /**
     * @param $host
     * @param int $port
     */
    public function setHost($host, $port = -1)
    {
        if ($port > 0) {
            $this->_port = $port;
        }

        $this->_host = $host;
    }

    /**
     * @return string
     */
    public function getUri()
    {
        $proto='http';
        if ($this->settings()->isHttps()) $proto='https';

        return $proto.'://' . $this->_host . ':' . $this->_port;
    }

    /**
     * @return Settings
     */
    public function settings()
    {
        return $this->_settings;
    }

    /**
     * @param $flag
     * @return mixed
     */
    public function verbose($flag)
    {
        $this->_verbose = $flag;
        return $flag;
    }

    /**
     * @param array $params
     * @return string
     */
    private function getUrl($params = [])
    {
        $settings = $this->settings()->getSettings();

        if (is_array($params) && sizeof($params)) {
            $settings = array_merge($settings, $params);
        }


        if ($this->settings()->isReadOnlyUser())
        {
            unset($settings['extremes']);
            unset($settings['readonly']);
            unset($settings['enable_http_compression']);
            unset($settings['max_execution_time']);

        }

        unset($settings['https']);


        return $this->getUri() . '?' . http_build_query($settings);
    }

    /**
     * @param $extendinfo
     * @return CurlerRequest
     */
    private function newRequest($extendinfo)
    {
        $new = new CurlerRequest();
        $new->auth($this->_username, $this->_password)
            ->POST()
            ->setRequestExtendedInfo($extendinfo);

        if ($this->settings()->isEnableHttpCompression()) {
            $new->httpCompression(true);
        }
        if ($this->settings()->getSessionId())
        {
            $new->persistent();
        }

        $new->timeOut($this->settings()->getTimeOut());
        $new->connectTimeOut($this->_connectTimeOut)->keepAlive();// one sec
        $new->verbose($this->_verbose);

        return $new;
    }

    /**
     * @param Query $query
     * @param array $urlParams
     * @param bool $query_as_string
     * @return CurlerRequest
     * @throws \ClickHouseDB\Exception\TransportException
     */
    private function makeRequest(Query $query, $urlParams = [], $query_as_string = false)
    {
        $sql = $query->toSql();

        if ($query_as_string) {
            $urlParams['query'] = $sql;
        }

        $url = $this->getUrl($urlParams);

        $extendinfo = [
            'sql' => $sql,
            'query' => $query,
            'format'=> $query->getFormat()
        ];

        $new = $this->newRequest($extendinfo);
        $new->url($url);




        if (!$query_as_string) {
            $new->parameters_json($sql);
        }
        if ($this->settings()->isEnableHttpCompression()) {
            $new->httpCompression(true);
        }

        return $new;
    }

    /**
     * @param $sql
     * @return CurlerRequest
     */
    public function writeStreamData($sql)
    {
        $query = new Query($sql);

        $url = $this->getUrl([
            'readonly' => 0,
            'query' => $query->toSql()
        ]);

        $extendinfo = [
            'sql' => $sql,
            'query' => $query,
            'format'=> $query->getFormat()
        ];

        $request = $this->newRequest($extendinfo);
        $request->url($url);
        return $request;
    }


    /**
     * @param $sql
     * @param $file_name
     * @return Statement
     * @throws \ClickHouseDB\Exception\TransportException
     */
    public function writeAsyncCSV($sql, $file_name)
    {
        $query = new Query($sql);

        $url = $this->getUrl([
            'readonly' => 0,
            'query' => $query->toSql()
        ]);

        $extendinfo = [
            'sql' => $sql,
            'query' => $query,
            'format'=> $query->getFormat()
        ];

        $request = $this->newRequest($extendinfo);
        $request->url($url);

        $request->setCallbackFunction(function (CurlerRequest $request) {
            fclose($request->getInfileHandle());

$request->getInfileHandle() of type boolean is incompatible with the type resource expected by parameter $handle of fclose().

        });

        $request->setInfile($file_name);
        $this->_curler->addQueLoop($request);

        return new Statement($request);
    }

    /**
     * get Count Pending Query in Queue
     *
     * @return int
     */
    public function getCountPendingQueue()
    {
        return $this->_curler->countPending();
    }

    /**
     * set Connect TimeOut in seconds [CURLOPT_CONNECTTIMEOUT] ( int )
     *
     * @param int $connectTimeOut
     */
    public function setConnectTimeOut($connectTimeOut)
    {
        $this->_connectTimeOut = $connectTimeOut;
    }

    /**
     * get ConnectTimeOut in seconds
     *
     * @return int
     */
    public function getConnectTimeOut()
    {
        return $this->_connectTimeOut;
    }


    public function __findXClickHouseProgress($handle)
    {
        $code=curl_getinfo($handle,CURLINFO_HTTP_CODE);

        // Search X-ClickHouse-Progress
        if ($code==200) {
            $response = curl_multi_getcontent($handle);
            $header_size = curl_getinfo($handle, CURLINFO_HEADER_SIZE);
            if (!$header_size) return false;

            $header = substr($response, 0, $header_size);
            if (!$header_size) return false;
            $pos=strrpos($header,'X-ClickHouse-Progress');

            if (!$pos) return false;

            $last=substr($header,$pos);
            $data=@json_decode(str_ireplace('X-ClickHouse-Progress:','',$last),true);

            if ($data && is_callable($this->xClickHouseProgress)) {

                if (is_array($this->xClickHouseProgress)){

The condition is_array($this->xClickHouseProgress) is always false.

                    call_user_func_array($this->xClickHouseProgress,[$data]);
                } else {
                    call_user_func($this->xClickHouseProgress,$data);

$this->xClickHouseProgress of type boolean is incompatible with the type callable expected by parameter $function of call_user_func().

                }


            }

        }

    }

    /**
     * @param Query $query
     * @param null $whereInFile

Are you sure the doc-type for parameter $writeToFile is correct as it would always require null to be passed?

Are you sure the doc-type for parameter $whereInFile is correct as it would always require null to be passed?

     * @param null $writeToFile
     * @return CurlerRequest
     * @throws \Exception
     */
    public function getRequestRead(Query $query, $whereInFile = null, $writeToFile = null)
    {
        $urlParams = ['readonly' => 1];
        $query_as_string = false;
        // ---------------------------------------------------------------------------------
        if ($whereInFile instanceof WhereInFile && $whereInFile->size()) {
            // $request = $this->prepareSelectWhereIn($request, $whereInFile);
            $structure = $whereInFile->fetchUrlParams();
            // $structure = [];
            $urlParams = array_merge($urlParams, $structure);
            $query_as_string = true;
        }
        // ---------------------------------------------------------------------------------
        // if result to file
        if ($writeToFile instanceof WriteToFile && $writeToFile->fetchFormat()) {
            $query->setFormat($writeToFile->fetchFormat());
            unset($urlParams['extremes']);
        }
        // ---------------------------------------------------------------------------------
        // makeRequest read
        $request = $this->makeRequest($query, $urlParams, $query_as_string);
        // ---------------------------------------------------------------------------------
        // attach files
        if ($whereInFile instanceof WhereInFile && $whereInFile->size()) {
            $request->attachFiles($whereInFile->fetchFiles());
        }
        // ---------------------------------------------------------------------------------
        // result to file
        if ($writeToFile instanceof WriteToFile && $writeToFile->fetchFormat()) {

            $fout = fopen($writeToFile->fetchFile(), 'w');
            $isGz = $writeToFile->getGzip();

            if ($isGz) {
                // write gzip header
                // "\x1f\x8b\x08\x00\x00\x00\x00\x00"
                // fwrite($fout, "\x1F\x8B\x08\x08".pack("V", time())."\0\xFF", 10);
                // write the original file name
                // $oname = str_replace("\0", "", basename($writeToFile->fetchFile()));
                // fwrite($fout, $oname."\0", 1+strlen($oname));

                fwrite($fout, "\x1f\x8b\x08\x00\x00\x00\x00\x00");

It seems like $fout can also be of type false; however, parameter $handle of fwrite() does only seem to accept resource, maybe add an additional type check?

            }


            $request->setResultFileHandle($fout, $isGz)->setCallbackFunction(function (CurlerRequest $request) {
                fclose($request->getResultFileHandle());
            });
        }
        if ($this->xClickHouseProgress)
        {
            $request->setFunctionProgress([$this,'__findXClickHouseProgress']);
        }
        // ---------------------------------------------------------------------------------
        return $request;

    }

    public function cleanQueryDegeneration()
    {
        $this->_query_degenerations = [];
        return true;
    }

    public function addQueryDegeneration(Degeneration $degeneration)
    {
        $this->_query_degenerations[] = $degeneration;
        return true;
    }

    /**
     * @param Query $query
     * @return CurlerRequest
     */
    public function getRequestWrite(Query $query)
    {
        $urlParams = ['readonly' => 0];
        return $this->makeRequest($query, $urlParams);
    }

    /**
     * @param $sql
     * @param $bindings
     * @return Query
     */
    private function prepareQuery($sql, $bindings)
    {

        // add Degeneration query
        foreach ($this->_query_degenerations as $degeneration) {
            $degeneration->bindParams($bindings);
        }

        return new Query($sql, $this->_query_degenerations);
    }


    /**
     * @param $sql
     * @param $bindings
     * @param $whereInFile
     * @param null $writeToFile

Are you sure the doc-type for parameter $writeToFile is correct as it would always require null to be passed?

     * @return CurlerRequest
     * @throws \Exception
     */
    private function prepareSelect($sql, $bindings, $whereInFile, $writeToFile = null)
    {
        if ($sql instanceof Query) {
            return $this->getRequestWrite($sql);
        }


        $query = $this->prepareQuery($sql, $bindings);
        $query->setFormat('JSON');
        return $this->getRequestRead($query, $whereInFile, $writeToFile);

    }

    /**
     * @param $sql
     * @param $bindings
     * @return CurlerRequest
     */
    private function prepareWrite($sql, $bindings = [])
    {
        if ($sql instanceof Query) {
            return $this->getRequestWrite($sql);
        }

        $query = $this->prepareQuery($sql, $bindings);
        return $this->getRequestWrite($query);
    }

    /**
     *
     * @return bool
     */
    public function executeAsync()
    {
        return $this->_curler->execLoopWait();
    }

    /**
     * @param $sql
     * @param array $bindings
     * @param null|WhereInFile $whereInFile
     * @param null|WriteToFile $writeToFile
     * @return Statement
     * @throws \ClickHouseDB\Exception\TransportException
     * @throws \Exception
     */
    public function select($sql, array $bindings = [], $whereInFile = null, $writeToFile = null)
    {
        $request = $this->prepareSelect($sql, $bindings, $whereInFile, $writeToFile);

It seems like $writeToFile can also be of type ClickHouseDB\Query\WriteToFile; however, parameter $writeToFile of ClickHouseDB\Transport\Http::prepareSelect() does only seem to accept null, maybe add an additional type check?

        $this->_curler->execOne($request);
        return new Statement($request);
    }

    /**
     * @param $sql
     * @param array $bindings
     * @param null $whereInFile

Are you sure the doc-type for parameter $whereInFile is correct as it would always require null to be passed?

Are you sure the doc-type for parameter $writeToFile is correct as it would always require null to be passed?

     * @param null $writeToFile
     * @return Statement
     * @throws \ClickHouseDB\Exception\TransportException
     * @throws \Exception
     */
    public function selectAsync($sql, array $bindings = [], $whereInFile = null, $writeToFile = null)
    {
        $request = $this->prepareSelect($sql, $bindings, $whereInFile, $writeToFile);
        $this->_curler->addQueLoop($request);
        return new Statement($request);
    }

    /**
     * @param $callback
     */
    public function setProgressFunction(callable $callback)
    {
        $this->xClickHouseProgress=$callback;

It seems like $callback of type callable is incompatible with the declared type boolean of property $xClickHouseProgress.

    }

    /**
     * @param $sql
     * @param array $bindings
     * @param bool $exception
     * @return Statement
     * @throws \ClickHouseDB\Exception\TransportException
     */
    public function write($sql, array $bindings = [], $exception = true)
    {
        $request = $this->prepareWrite($sql, $bindings);
        $this->_curler->execOne($request);
        $response = new Statement($request);
        if ($exception) {
            if ($response->isError()) {
                $response->error();
            }
        }
        return $response;
    }
}