Можно ли ожидать, что пул Guzzle будет ждать запросов?
Прямо сейчас я могу добавлять запросы в пул динамически, но как только пул пуст, жужжание остановится (очевидно).
Это проблема, когда я делаю 10 или около того страниц одновременно, потому что мой массив запросов будет пустым, пока не будут обработаны результирующие HTML-страницы и добавлены новые ссылки.
Это мой генератор:
$generator = function () { while ($request = array_shift($this->requests)) { if (isset($request['page'])) { $key = 'page_' . $request['page']; } else { $key = 'listing_' . $request['listing']; } yield $key => new Request('GET', $request['url']); } echo "Exiting...\n"; flush(); }; И мой бассейн:
$pool = new Pool($this->client, $generator(), [ 'concurrency' => function() { return max(1, min(count($this->requests), 2)); }, 'fulfilled' => function ($response, $index) { // new requests may be added to the $this->requests array here } //... ]); $promise = $pool->promise(); $promise->wait();
Отредактированный код после ответа от @Alexey Shockov:
$generator = function() use ($headers) { while ($request = array_shift($this->requests)) { echo 'Requesting ' . $request['id'] . ': ' . $request['url'] . "\r\n"; $r = new Request('GET', $request['url'], $headers); yield 'id_' . $request['id'] => $this->client->sendAsync($r)->then(function($response, $index) { echo 'In promise fulfillment ' . $index . "\r\n"; }, function($reason, $index) { echo 'in rejected: ' . $index . "\r\n"; }); } }; $promise = \GuzzleHttp\Promise\each_limit($generator(), 10, function() { echo 'fullfilled' . "\r\n"; flush(); }, function($err) { echo 'rejected' . "\r\n"; echo $err->getMessage(); flush(); }); $promise->wait();
К сожалению, вы не можете сделать это с генератором, только с помощью пользовательского итератора.
Я подготовил суть с полным примером , но основная идея состоит в том, чтобы создать Итератор, который изменит свое состояние в обоих направлениях (он может снова стать действительным после завершения).
Пример с ArrayIterator в psysh :
>>> $a = new ArrayIterator([1, 2]) => ArrayIterator {#186 +0: 1, +1: 2, } >>> $a->current() => 1 >>> $a->next() => null >>> $a->current() => 2 >>> $a->next() => null >>> $a->valid() => false >>> $a[] = 2 => 2 >>> $a->valid() => true >>> $a->current() => 2
Имея в виду эту идею, мы можем передать такой динамический итератор в Guzzle и оставить это делать:
// MapIterator mainly needed for readability. $generator = new MapIterator( // Initial data. This object will be always passed as the second parameter to the callback below new \ArrayIterator(['http://google.com']), function ($request, $array) use ($httpClient, $next) { return $httpClient->requestAsync('GET', $request) ->then(function (Response $response) use ($request, $array, $next) { // The status code for example. echo $request . ': ' . $response->getStatusCode() . PHP_EOL; // New requests. $array->append($next->shift()); $array->append($next->shift()); }); } ); // The "magic". $generator = new ExpectingIterator($generator); // And the concurrent runner. $promise = \GuzzleHttp\Promise\each_limit($generator, 5); $promise->wait();
Как я уже говорил, полный пример в основном MapIterator с MapIterator и ExpectingIterator .
По-видимому, из вопроса, что вы можете переместить обратный вызов агрегации непосредственно в запрос. В этом случае пул будет всегда ждать вашего кода обработки, поэтому вы сможете добавлять новые запросы в любой момент.
Генератор может возвращать либо запрос, либо обещание, и обещания могут быть объединены по-разному.
$generator = function () { while ($request = array_shift($this->requests)) { if (isset($request['page'])) { $key = 'page_' . $request['page']; } else { $key = 'listing_' . $request['listing']; } yield $this->client->sendAsync('GET', $request['url']) ->then(function (Response $response) use ($key) { /* * The fullfillment callback is now connected to the query, so the * pool will wait for it. * * $key is also available, because it's just a closure, so * no $index needed as an argument. */ }); } echo "Exiting...\n"; flush(); }; $promise = \GuzzleHttp\Promise\each_limit($generator(), [ 'concurrency' => function () { return max(1, min(count($this->requests), 2)); }, //... ]); $promise->wait();
Как я уже говорил, полный пример в основном связан с MapIterator и ExpectingIterator
Итераторы снова не вернутся к php <7, ваш пример с arrayIterator и образец с MapIterator останавливаются после исчерпания первоначального пула …
С другой стороны, все это работает на более ранних версиях php, если вы используете -> append method на итераторе вместо [] push.
Ответ – да, вы можете. Вам просто нужно больше генераторов. И разделить логику запроса и логику очередей на асинхронный дизайн. Вместо того, чтобы использовать массив для запросов, которые ваш пул собирается выдавать, и ждать его, должен быть сам генератор, который дает новые запросы из вашего первоначального списка и запросы, добавленные из разобранных ответов, до тех пор, пока все запросы не будут отправлены разобранным, а результирующий запрос будет отправлен и встречается (повторяющееся) или условие остановки.