8131173364

Committed 03 Mar 2024 03:18PM UTC coverage: 82.576% (-1.7%) from 84.298%

Build # 8131173364

Build Type

push

github

Committed by

Smoren

Commit Message

phpstan fixes

Run Details

6 of 11 new or added lines in 1 file covered. (54.55%)

5 existing lines in 1 file now uncovered.

109 of 132 relevant lines covered (82.58%)

0.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.23

/src/MultiCurlRunner.php

<?php

namespace Smoren\MultiCurl;

use CurlHandle;
use CurlMultiHandle;
use RuntimeException;

/**
 * Class to run MultiCurl requests and get responses
 * @author <ofigate@gmail.com> Smoren
 */
class MultiCurlRunner
{
    /**
     * @var resource|CurlMultiHandle MultiCurl resource
     */
    protected $mh;
    /**
     * @var array<int, string> map [workerId => customRequestId]
     */
    protected $workersMap;
    /**
     * @var array<resource> unemployed workers stack
     */
    protected $unemployedWorkers;
    /**
     * @var int max parallel connections count
     */
    protected $maxConnections;
    /**
     * @var array<string, array<int, mixed>> map of CURL options including headers by custom request ID
     */
    protected $requestsConfigMap;
    /**
     * @var array<string, array<string, mixed>> responses mapped by custom request ID
     */
    protected $result;

    /**
     * MultiCurlRunner constructor
     * @param array<string, array<int, mixed>> $requestsConfigMap map of CURL options
     * including headers by custom request ID
     * @param int $maxConnections max parallel connections count
     */
    public function __construct(array $requestsConfigMap, int $maxConnections)
    {
        $this->requestsConfigMap = $requestsConfigMap;
        $this->maxConnections = min($maxConnections, count($requestsConfigMap));

        $mh = curl_multi_init();

        $this->mh = $mh;
        $this->workersMap = [];
        $this->unemployedWorkers = [];
        $this->result = [];
    }

    /**
     * Makes requests and stores responses
     * @return self
     * @throws RuntimeException
     */
    public function run(): self
    {
        for($i=0; $i<$this->maxConnections; ++$i) {
            /** @var resource|false $unemployedWorker */
            $unemployedWorker = curl_init();
            if(!$unemployedWorker) {
                throw new RuntimeException("failed creating unemployed worker #{$i}");
            }
            $this->unemployedWorkers[] = $unemployedWorker;
        }
        unset($i, $this->unemployedWorker);

        foreach($this->requestsConfigMap as $id => $options) {
            while(!count($this->unemployedWorkers)) {
                $this->doWork();
            }

            $options[CURLOPT_HEADER] = 1;

            $newWorker = array_pop($this->unemployedWorkers);

            // @phpstan-ignore-next-line
            if(!curl_setopt_array($newWorker, $options)) {
                $errNo = curl_errno($newWorker); // @phpstan-ignore-line
                $errMess = curl_error($newWorker); // @phpstan-ignore-line
                $errData = var_export($options, true);
                throw new RuntimeException("curl_setopt_array failed: {$errNo} {$errMess} {$errData}");
            }

            $this->workersMap[(int)$newWorker] = $id;
            curl_multi_add_handle($this->mh, $newWorker); // @phpstan-ignore-line
        }
        unset($options);

        while(count($this->workersMap) > 0) {
            $this->doWork();
        }

        foreach($this->unemployedWorkers as $unemployedWorker) {
            curl_close($unemployedWorker); // @phpstan-ignore-line
        }

        curl_multi_close($this->mh); // @phpstan-ignore-line

        return $this;
    }

    /**
     * Returns response:
     * [customRequestId => [code => statusCode, headers => [key => value, ...], body => responseBody], ...]
     * @param bool $okOnly if true: return only responses with (200 <= status code < 300)
     * @return array<string, array<string, mixed>> responses mapped by custom request IDs
     */
    public function getResult(bool $okOnly = false): array
    {
        $result = [];

        foreach($this->result as $key => $value) {
            if(!$okOnly || $value['code'] === 200) {
                $result[$key] = $value;
            }
        }

        return $result;
    }

    /**
     * Returns response bodies:
     * [customRequestId => responseBody, ...]
     * @param bool $okOnly if true: return only responses with (200 <= status code < 300)
     * @return array<string, mixed> responses mapped by custom request IDs
     */
    public function getResultData(bool $okOnly = false): array
    {
        $result = [];

        foreach($this->result as $key => $value) {
            if(!$okOnly || $value['code'] >= 200 && $value['code'] < 300) {
                $result[$key] = $value['body'];
            }
        }

        return $result;
    }

    /**
     * Manages workers during making the requests
     * @return void
     */
    protected function doWork(): void
    {
        assert(count($this->workersMap) > 0, "work() called with 0 workers!!");
        $stillRunning = null;

        while(true) {
            do {
                $err = curl_multi_exec($this->mh, $stillRunning); // @phpstan-ignore-line
            } while($err === CURLM_CALL_MULTI_PERFORM);

            if($err !== CURLM_OK) {
                $errInfo = [
                    "multi_exec_return" => $err,
                    "curl_multi_errno" => curl_multi_errno($this->mh), // @phpstan-ignore-line
                    "curl_multi_strerror" => curl_multi_strerror($err)
                ];

                $errData = str_replace(["\r", "\n"], "", var_export($errInfo, true));
                throw new RuntimeException("curl_multi_exec error: {$errData}");
            }
            if($stillRunning < count($this->workersMap)) {
                // some workers has finished downloading, process them
                // echo "processing!";
                break;
            } else {
                // no workers finished yet, sleep-wait for workers to finish downloading.
                curl_multi_select($this->mh, 1); // @phpstan-ignore-line
                // sleep(1);
            }
        }
        // @phpstan-ignore-next-line
        while(($info = curl_multi_info_read($this->mh)) !== false) {
            if($info['msg'] !== CURLMSG_DONE) {
                // no idea what this is, it's not the message we're looking for though, ignore it.
                continue;
            }

            if($info['result'] !== CURLM_OK) {
                $errInfo = [
                    "effective_url" => curl_getinfo($info['handle'], CURLINFO_EFFECTIVE_URL),
                    "curl_errno" => curl_errno($info['handle']),
                    "curl_error" => curl_error($info['handle']),
                    "curl_multi_errno" => curl_multi_errno($this->mh), // @phpstan-ignore-line
                    "curl_multi_strerror" => curl_multi_strerror(curl_multi_errno($this->mh)) // @phpstan-ignore-line
                ];

                $errData = str_replace(["\r", "\n"], "", var_export($errInfo, true));
                throw new RuntimeException("curl_multi worker error: {$errData}");
            }

            $ch = $info['handle'];
            $chIndex = (int)$ch;

            // @phpstan-ignore-next-line
            $this->result[$this->workersMap[$chIndex]] = $this->parseResponse(curl_multi_getcontent($ch));

            unset($this->workersMap[$chIndex]);
            curl_multi_remove_handle($this->mh, $ch); // @phpstan-ignore-line
            $this->unemployedWorkers[] = $ch;
        }
    }

    /**
     * Parses the response
     * @param string $response raw HTTP response
     * @return array<string, mixed> [code => statusCode, headers => [key => value, ...], body => responseBody]
     */
    protected function parseResponse(string $response): array
    {
        $arResponse = explode("\r\n\r\n", $response);

        $arHeaders = [];
        $statusCode = null;
        $body = null;

        while(count($arResponse)) {
            $respItem = array_shift($arResponse);

            $line = (string)strtok($respItem, "\r\n");
            $statusCodeLine = trim($line);
            if(preg_match('|HTTP/[\d.]+\s+(\d+)|', $statusCodeLine, $matches)) {
                $arHeaders = [];

                if(isset($matches[1])) {
                    $statusCode = (int)$matches[1];
                } else {
                    $statusCode = null;
                }

                // Parse the string, saving it into an array instead
                while(($line = strtok("\r\n")) !== false) {
                    if(($matches = explode(':', $line, 2)) !== false) {
                        $arHeaders[trim(mb_strtolower($matches[0]))] = trim(mb_strtolower($matches[1]));
                    }
                }
            } else {
                $contentType = $arHeaders['content-type'] ?? null;
                if($contentType === 'application/json') {
                    $body = json_decode($respItem, true);
                } else {
                    $body = $respItem;
                }
            }
        }

        return [
            'code' => $statusCode,
            'headers' => $arHeaders,
            'body' => $body,
        ];
    }
}

1	<?php
2
3	namespace Smoren\MultiCurl;
4
5	use CurlHandle;
6	use CurlMultiHandle;
7	use RuntimeException;
8
9	/**
10	* Class to run MultiCurl requests and get responses
11	* @author <ofigate@gmail.com> Smoren
12	*/
13	class MultiCurlRunner
14	{
15	/**
16	* @var resource\|CurlMultiHandle MultiCurl resource
17	*/
18	protected $mh;
19	/**
20	* @var array<int, string> map [workerId => customRequestId]
21	*/
22	protected $workersMap;
23	/**
24	* @var array<resource> unemployed workers stack
25	*/
26	protected $unemployedWorkers;
27	/**
28	* @var int max parallel connections count
29	*/
30	protected $maxConnections;
31	/**
32	* @var array<string, array<int, mixed>> map of CURL options including headers by custom request ID
33	*/
34	protected $requestsConfigMap;
35	/**
36	* @var array<string, array<string, mixed>> responses mapped by custom request ID
37	*/
38	protected $result;
39
40	/**
41	* MultiCurlRunner constructor
42	* @param array<string, array<int, mixed>> $requestsConfigMap map of CURL options
43	* including headers by custom request ID
44	* @param int $maxConnections max parallel connections count
45	*/
46	public function __construct(array $requestsConfigMap, int $maxConnections)
47	{
48	$this->requestsConfigMap = $requestsConfigMap;	1✔
49	$this->maxConnections = min($maxConnections, count($requestsConfigMap));	1✔
50
51	$mh = curl_multi_init();	1✔
52
53	$this->mh = $mh;	1✔
54	$this->workersMap = [];	1✔
55	$this->unemployedWorkers = [];	1✔
56	$this->result = [];	1✔
57	}
58
59	/**
60	* Makes requests and stores responses
61	* @return self
62	* @throws RuntimeException
63	*/
64	public function run(): self
65	{
66	for($i=0; $i<$this->maxConnections; ++$i) {	1✔
67	/** @var resource\|false $unemployedWorker */
68	$unemployedWorker = curl_init();	1✔
69	if(!$unemployedWorker) {	1✔
70	throw new RuntimeException("failed creating unemployed worker #{$i}");	×
71	}
72	$this->unemployedWorkers[] = $unemployedWorker;	1✔
73	}
74	unset($i, $this->unemployedWorker);	1✔
75
76	foreach($this->requestsConfigMap as $id => $options) {	1✔
77	while(!count($this->unemployedWorkers)) {	1✔
78	$this->doWork();	1✔
79	}
80
81	$options[CURLOPT_HEADER] = 1;	1✔
82
83	$newWorker = array_pop($this->unemployedWorkers);	1✔
84
85	// @phpstan-ignore-next-line
86	if(!curl_setopt_array($newWorker, $options)) {	1✔
NEW 87	$errNo = curl_errno($newWorker); // @phpstan-ignore-line	×
NEW 88	$errMess = curl_error($newWorker); // @phpstan-ignore-line	×
89	$errData = var_export($options, true);	×
90	throw new RuntimeException("curl_setopt_array failed: {$errNo} {$errMess} {$errData}");	×
91	}
92
93	$this->workersMap[(int)$newWorker] = $id;	1✔
94	curl_multi_add_handle($this->mh, $newWorker); // @phpstan-ignore-line	1✔
95	}
96	unset($options);	1✔
97
98	while(count($this->workersMap) > 0) {	1✔
99	$this->doWork();	1✔
100	}
101
102	foreach($this->unemployedWorkers as $unemployedWorker) {	1✔
103	curl_close($unemployedWorker); // @phpstan-ignore-line	1✔
104	}
105
106	curl_multi_close($this->mh); // @phpstan-ignore-line	1✔
107
108	return $this;	1✔
109	}
110
111	/**
112	* Returns response:
113	* [customRequestId => [code => statusCode, headers => [key => value, ...], body => responseBody], ...]
114	* @param bool $okOnly if true: return only responses with (200 <= status code < 300)
115	* @return array<string, array<string, mixed>> responses mapped by custom request IDs
116	*/
117	public function getResult(bool $okOnly = false): array
118	{
119	$result = [];	1✔
120
121	foreach($this->result as $key => $value) {	1✔
122	if(!$okOnly \|\| $value['code'] === 200) {	1✔
123	$result[$key] = $value;	1✔
124	}
125	}
126
127	return $result;	1✔
128	}
129
130	/**
131	* Returns response bodies:
132	* [customRequestId => responseBody, ...]
133	* @param bool $okOnly if true: return only responses with (200 <= status code < 300)
134	* @return array<string, mixed> responses mapped by custom request IDs
135	*/
136	public function getResultData(bool $okOnly = false): array
137	{
138	$result = [];	1✔
139
140	foreach($this->result as $key => $value) {	1✔
141	if(!$okOnly \|\| $value['code'] >= 200 && $value['code'] < 300) {	1✔
142	$result[$key] = $value['body'];	1✔
143	}
144	}
145
146	return $result;	1✔
147	}
148
149	/**
150	* Manages workers during making the requests
151	* @return void
152	*/
153	protected function doWork(): void
154	{
155	assert(count($this->workersMap) > 0, "work() called with 0 workers!!");
156	$stillRunning = null;	1✔
157
158	while(true) {	1✔
159	do {
160	$err = curl_multi_exec($this->mh, $stillRunning); // @phpstan-ignore-line	1✔
161	} while($err === CURLM_CALL_MULTI_PERFORM);	1✔
162
163	if($err !== CURLM_OK) {	1✔
UNCOV 164	$errInfo = [	×
UNCOV 165	"multi_exec_return" => $err,	×
NEW 166	"curl_multi_errno" => curl_multi_errno($this->mh), // @phpstan-ignore-line	×
167	"curl_multi_strerror" => curl_multi_strerror($err)	×
UNCOV 168	];	×
169
170	$errData = str_replace(["\r", "\n"], "", var_export($errInfo, true));	×
171	throw new RuntimeException("curl_multi_exec error: {$errData}");	×
172	}
173	if($stillRunning < count($this->workersMap)) {	1✔
174	// some workers has finished downloading, process them
175	// echo "processing!";
176	break;	1✔
177	} else {
178	// no workers finished yet, sleep-wait for workers to finish downloading.
179	curl_multi_select($this->mh, 1); // @phpstan-ignore-line	1✔
180	// sleep(1);
181	}
182	}
183	// @phpstan-ignore-next-line
184	while(($info = curl_multi_info_read($this->mh)) !== false) {	1✔
185	if($info['msg'] !== CURLMSG_DONE) {	1✔
186	// no idea what this is, it's not the message we're looking for though, ignore it.
187	continue;	×
188	}
189
190	if($info['result'] !== CURLM_OK) {	1✔
UNCOV 191	$errInfo = [	×
192	"effective_url" => curl_getinfo($info['handle'], CURLINFO_EFFECTIVE_URL),	×
193	"curl_errno" => curl_errno($info['handle']),	×
194	"curl_error" => curl_error($info['handle']),	×
NEW 195	"curl_multi_errno" => curl_multi_errno($this->mh), // @phpstan-ignore-line	×
NEW 196	"curl_multi_strerror" => curl_multi_strerror(curl_multi_errno($this->mh)) // @phpstan-ignore-line	×
UNCOV 197	];	×
198
199	$errData = str_replace(["\r", "\n"], "", var_export($errInfo, true));	×
200	throw new RuntimeException("curl_multi worker error: {$errData}");	×
201	}
202
203	$ch = $info['handle'];	1✔
204	$chIndex = (int)$ch;	1✔
205
206	// @phpstan-ignore-next-line
207	$this->result[$this->workersMap[$chIndex]] = $this->parseResponse(curl_multi_getcontent($ch));	1✔
208
209	unset($this->workersMap[$chIndex]);	1✔
210	curl_multi_remove_handle($this->mh, $ch); // @phpstan-ignore-line	1✔
211	$this->unemployedWorkers[] = $ch;	1✔
212	}
213	}
214
215	/**
216	* Parses the response
217	* @param string $response raw HTTP response
218	* @return array<string, mixed> [code => statusCode, headers => [key => value, ...], body => responseBody]
219	*/
220	protected function parseResponse(string $response): array
221	{
222	$arResponse = explode("\r\n\r\n", $response);	1✔
223
224	$arHeaders = [];	1✔
225	$statusCode = null;	1✔
226	$body = null;	1✔
227
228	while(count($arResponse)) {	1✔
229	$respItem = array_shift($arResponse);	1✔
230
231	$line = (string)strtok($respItem, "\r\n");	1✔
232	$statusCodeLine = trim($line);	1✔
233	if(preg_match('\|HTTP/[\d.]+\s+(\d+)\|', $statusCodeLine, $matches)) {	1✔
234	$arHeaders = [];	1✔
235
236	if(isset($matches[1])) {	1✔
237	$statusCode = (int)$matches[1];	1✔
238	} else {
239	$statusCode = null;	×
240	}
241
242	// Parse the string, saving it into an array instead
243	while(($line = strtok("\r\n")) !== false) {	1✔
244	if(($matches = explode(':', $line, 2)) !== false) {	1✔
245	$arHeaders[trim(mb_strtolower($matches[0]))] = trim(mb_strtolower($matches[1]));	1✔
246	}
247	}
248	} else {
249	$contentType = $arHeaders['content-type'] ?? null;	1✔
250	if($contentType === 'application/json') {	1✔
251	$body = json_decode($respItem, true);	1✔
252	} else {
253	$body = $respItem;	1✔
254	}
255	}
256	}
257
258	return [	1✔
259	'code' => $statusCode,	1✔
260	'headers' => $arHeaders,	1✔
261	'body' => $body,	1✔
262	];	1✔
263	}
264	}

Smoren / multicurl-php / 8131173364

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous