1 | <?php
|
---|
2 |
|
---|
3 | namespace GuzzleHttp\Handler;
|
---|
4 |
|
---|
5 | use Closure;
|
---|
6 | use GuzzleHttp\Promise as P;
|
---|
7 | use GuzzleHttp\Promise\Promise;
|
---|
8 | use GuzzleHttp\Promise\PromiseInterface;
|
---|
9 | use GuzzleHttp\Utils;
|
---|
10 | use Psr\Http\Message\RequestInterface;
|
---|
11 |
|
---|
12 | /**
|
---|
13 | * Returns an asynchronous response using curl_multi_* functions.
|
---|
14 | *
|
---|
15 | * When using the CurlMultiHandler, custom curl options can be specified as an
|
---|
16 | * associative array of curl option constants mapping to values in the
|
---|
17 | * **curl** key of the provided request options.
|
---|
18 | *
|
---|
19 | * @final
|
---|
20 | */
|
---|
21 | class CurlMultiHandler
|
---|
22 | {
|
---|
23 | /**
|
---|
24 | * @var CurlFactoryInterface
|
---|
25 | */
|
---|
26 | private $factory;
|
---|
27 |
|
---|
28 | /**
|
---|
29 | * @var int
|
---|
30 | */
|
---|
31 | private $selectTimeout;
|
---|
32 |
|
---|
33 | /**
|
---|
34 | * @var int Will be higher than 0 when `curl_multi_exec` is still running.
|
---|
35 | */
|
---|
36 | private $active = 0;
|
---|
37 |
|
---|
38 | /**
|
---|
39 | * @var array Request entry handles, indexed by handle id in `addRequest`.
|
---|
40 | *
|
---|
41 | * @see CurlMultiHandler::addRequest
|
---|
42 | */
|
---|
43 | private $handles = [];
|
---|
44 |
|
---|
45 | /**
|
---|
46 | * @var array<int, float> An array of delay times, indexed by handle id in `addRequest`.
|
---|
47 | *
|
---|
48 | * @see CurlMultiHandler::addRequest
|
---|
49 | */
|
---|
50 | private $delays = [];
|
---|
51 |
|
---|
52 | /**
|
---|
53 | * @var array<mixed> An associative array of CURLMOPT_* options and corresponding values for curl_multi_setopt()
|
---|
54 | */
|
---|
55 | private $options = [];
|
---|
56 |
|
---|
57 | /** @var resource|\CurlMultiHandle */
|
---|
58 | private $_mh;
|
---|
59 |
|
---|
60 | /**
|
---|
61 | * This handler accepts the following options:
|
---|
62 | *
|
---|
63 | * - handle_factory: An optional factory used to create curl handles
|
---|
64 | * - select_timeout: Optional timeout (in seconds) to block before timing
|
---|
65 | * out while selecting curl handles. Defaults to 1 second.
|
---|
66 | * - options: An associative array of CURLMOPT_* options and
|
---|
67 | * corresponding values for curl_multi_setopt()
|
---|
68 | */
|
---|
69 | public function __construct(array $options = [])
|
---|
70 | {
|
---|
71 | $this->factory = $options['handle_factory'] ?? new CurlFactory(50);
|
---|
72 |
|
---|
73 | if (isset($options['select_timeout'])) {
|
---|
74 | $this->selectTimeout = $options['select_timeout'];
|
---|
75 | } elseif ($selectTimeout = Utils::getenv('GUZZLE_CURL_SELECT_TIMEOUT')) {
|
---|
76 | @trigger_error('Since guzzlehttp/guzzle 7.2.0: Using environment variable GUZZLE_CURL_SELECT_TIMEOUT is deprecated. Use option "select_timeout" instead.', \E_USER_DEPRECATED);
|
---|
77 | $this->selectTimeout = (int) $selectTimeout;
|
---|
78 | } else {
|
---|
79 | $this->selectTimeout = 1;
|
---|
80 | }
|
---|
81 |
|
---|
82 | $this->options = $options['options'] ?? [];
|
---|
83 |
|
---|
84 | // unsetting the property forces the first access to go through
|
---|
85 | // __get().
|
---|
86 | unset($this->_mh);
|
---|
87 | }
|
---|
88 |
|
---|
89 | /**
|
---|
90 | * @param string $name
|
---|
91 | *
|
---|
92 | * @return resource|\CurlMultiHandle
|
---|
93 | *
|
---|
94 | * @throws \BadMethodCallException when another field as `_mh` will be gotten
|
---|
95 | * @throws \RuntimeException when curl can not initialize a multi handle
|
---|
96 | */
|
---|
97 | public function __get($name)
|
---|
98 | {
|
---|
99 | if ($name !== '_mh') {
|
---|
100 | throw new \BadMethodCallException("Can not get other property as '_mh'.");
|
---|
101 | }
|
---|
102 |
|
---|
103 | $multiHandle = \curl_multi_init();
|
---|
104 |
|
---|
105 | if (false === $multiHandle) {
|
---|
106 | throw new \RuntimeException('Can not initialize curl multi handle.');
|
---|
107 | }
|
---|
108 |
|
---|
109 | $this->_mh = $multiHandle;
|
---|
110 |
|
---|
111 | foreach ($this->options as $option => $value) {
|
---|
112 | // A warning is raised in case of a wrong option.
|
---|
113 | curl_multi_setopt($this->_mh, $option, $value);
|
---|
114 | }
|
---|
115 |
|
---|
116 | return $this->_mh;
|
---|
117 | }
|
---|
118 |
|
---|
119 | public function __destruct()
|
---|
120 | {
|
---|
121 | if (isset($this->_mh)) {
|
---|
122 | \curl_multi_close($this->_mh);
|
---|
123 | unset($this->_mh);
|
---|
124 | }
|
---|
125 | }
|
---|
126 |
|
---|
127 | public function __invoke(RequestInterface $request, array $options): PromiseInterface
|
---|
128 | {
|
---|
129 | $easy = $this->factory->create($request, $options);
|
---|
130 | $id = (int) $easy->handle;
|
---|
131 |
|
---|
132 | $promise = new Promise(
|
---|
133 | [$this, 'execute'],
|
---|
134 | function () use ($id) {
|
---|
135 | return $this->cancel($id);
|
---|
136 | }
|
---|
137 | );
|
---|
138 |
|
---|
139 | $this->addRequest(['easy' => $easy, 'deferred' => $promise]);
|
---|
140 |
|
---|
141 | return $promise;
|
---|
142 | }
|
---|
143 |
|
---|
144 | /**
|
---|
145 | * Ticks the curl event loop.
|
---|
146 | */
|
---|
147 | public function tick(): void
|
---|
148 | {
|
---|
149 | // Add any delayed handles if needed.
|
---|
150 | if ($this->delays) {
|
---|
151 | $currentTime = Utils::currentTime();
|
---|
152 | foreach ($this->delays as $id => $delay) {
|
---|
153 | if ($currentTime >= $delay) {
|
---|
154 | unset($this->delays[$id]);
|
---|
155 | \curl_multi_add_handle(
|
---|
156 | $this->_mh,
|
---|
157 | $this->handles[$id]['easy']->handle
|
---|
158 | );
|
---|
159 | }
|
---|
160 | }
|
---|
161 | }
|
---|
162 |
|
---|
163 | // Run curl_multi_exec in the queue to enable other async tasks to run
|
---|
164 | P\Utils::queue()->add(Closure::fromCallable([$this, 'tickInQueue']));
|
---|
165 |
|
---|
166 | // Step through the task queue which may add additional requests.
|
---|
167 | P\Utils::queue()->run();
|
---|
168 |
|
---|
169 | if ($this->active && \curl_multi_select($this->_mh, $this->selectTimeout) === -1) {
|
---|
170 | // Perform a usleep if a select returns -1.
|
---|
171 | // See: https://bugs.php.net/bug.php?id=61141
|
---|
172 | \usleep(250);
|
---|
173 | }
|
---|
174 |
|
---|
175 | while (\curl_multi_exec($this->_mh, $this->active) === \CURLM_CALL_MULTI_PERFORM) {
|
---|
176 | // Prevent busy looping for slow HTTP requests.
|
---|
177 | \curl_multi_select($this->_mh, $this->selectTimeout);
|
---|
178 | }
|
---|
179 |
|
---|
180 | $this->processMessages();
|
---|
181 | }
|
---|
182 |
|
---|
183 | /**
|
---|
184 | * Runs \curl_multi_exec() inside the event loop, to prevent busy looping
|
---|
185 | */
|
---|
186 | private function tickInQueue(): void
|
---|
187 | {
|
---|
188 | if (\curl_multi_exec($this->_mh, $this->active) === \CURLM_CALL_MULTI_PERFORM) {
|
---|
189 | \curl_multi_select($this->_mh, 0);
|
---|
190 | P\Utils::queue()->add(Closure::fromCallable([$this, 'tickInQueue']));
|
---|
191 | }
|
---|
192 | }
|
---|
193 |
|
---|
194 | /**
|
---|
195 | * Runs until all outstanding connections have completed.
|
---|
196 | */
|
---|
197 | public function execute(): void
|
---|
198 | {
|
---|
199 | $queue = P\Utils::queue();
|
---|
200 |
|
---|
201 | while ($this->handles || !$queue->isEmpty()) {
|
---|
202 | // If there are no transfers, then sleep for the next delay
|
---|
203 | if (!$this->active && $this->delays) {
|
---|
204 | \usleep($this->timeToNext());
|
---|
205 | }
|
---|
206 | $this->tick();
|
---|
207 | }
|
---|
208 | }
|
---|
209 |
|
---|
210 | private function addRequest(array $entry): void
|
---|
211 | {
|
---|
212 | $easy = $entry['easy'];
|
---|
213 | $id = (int) $easy->handle;
|
---|
214 | $this->handles[$id] = $entry;
|
---|
215 | if (empty($easy->options['delay'])) {
|
---|
216 | \curl_multi_add_handle($this->_mh, $easy->handle);
|
---|
217 | } else {
|
---|
218 | $this->delays[$id] = Utils::currentTime() + ($easy->options['delay'] / 1000);
|
---|
219 | }
|
---|
220 | }
|
---|
221 |
|
---|
222 | /**
|
---|
223 | * Cancels a handle from sending and removes references to it.
|
---|
224 | *
|
---|
225 | * @param int $id Handle ID to cancel and remove.
|
---|
226 | *
|
---|
227 | * @return bool True on success, false on failure.
|
---|
228 | */
|
---|
229 | private function cancel($id): bool
|
---|
230 | {
|
---|
231 | if (!is_int($id)) {
|
---|
232 | trigger_deprecation('guzzlehttp/guzzle', '7.4', 'Not passing an integer to %s::%s() is deprecated and will cause an error in 8.0.', __CLASS__, __FUNCTION__);
|
---|
233 | }
|
---|
234 |
|
---|
235 | // Cannot cancel if it has been processed.
|
---|
236 | if (!isset($this->handles[$id])) {
|
---|
237 | return false;
|
---|
238 | }
|
---|
239 |
|
---|
240 | $handle = $this->handles[$id]['easy']->handle;
|
---|
241 | unset($this->delays[$id], $this->handles[$id]);
|
---|
242 | \curl_multi_remove_handle($this->_mh, $handle);
|
---|
243 | \curl_close($handle);
|
---|
244 |
|
---|
245 | return true;
|
---|
246 | }
|
---|
247 |
|
---|
248 | private function processMessages(): void
|
---|
249 | {
|
---|
250 | while ($done = \curl_multi_info_read($this->_mh)) {
|
---|
251 | if ($done['msg'] !== \CURLMSG_DONE) {
|
---|
252 | // if it's not done, then it would be premature to remove the handle. ref https://github.com/guzzle/guzzle/pull/2892#issuecomment-945150216
|
---|
253 | continue;
|
---|
254 | }
|
---|
255 | $id = (int) $done['handle'];
|
---|
256 | \curl_multi_remove_handle($this->_mh, $done['handle']);
|
---|
257 |
|
---|
258 | if (!isset($this->handles[$id])) {
|
---|
259 | // Probably was cancelled.
|
---|
260 | continue;
|
---|
261 | }
|
---|
262 |
|
---|
263 | $entry = $this->handles[$id];
|
---|
264 | unset($this->handles[$id], $this->delays[$id]);
|
---|
265 | $entry['easy']->errno = $done['result'];
|
---|
266 | $entry['deferred']->resolve(
|
---|
267 | CurlFactory::finish($this, $entry['easy'], $this->factory)
|
---|
268 | );
|
---|
269 | }
|
---|
270 | }
|
---|
271 |
|
---|
272 | private function timeToNext(): int
|
---|
273 | {
|
---|
274 | $currentTime = Utils::currentTime();
|
---|
275 | $nextTime = \PHP_INT_MAX;
|
---|
276 | foreach ($this->delays as $time) {
|
---|
277 | if ($time < $nextTime) {
|
---|
278 | $nextTime = $time;
|
---|
279 | }
|
---|
280 | }
|
---|
281 |
|
---|
282 | return ((int) \max(0, $nextTime - $currentTime)) * 1000000;
|
---|
283 | }
|
---|
284 | }
|
---|