Бассейн Гузл: ждать запросов

Возможно ли, что пул 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();

Ответ 1

К сожалению, вы не можете сделать это с генератором, только с пользовательским итератором.

Я подготовил текст с полным примером, но основная идея - просто создать итератор, который изменит свое состояние в обоих направлениях (он может снова стать действительным после окончания).

Пример с 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 и ExpectingIterator.

Ответ 2

Кажется, из вопроса, что вы можете перемещать обратный вызов агрегации непосредственно в запрос. В этом случае пул будет всегда ждать вашего кода обработки, поэтому вы сможете добавлять новые запросы в любой момент.

Генератор может возвращать либо запрос, либо обещание, а promises можно комбинировать по-разному.

$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 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();

Ответ 3

Как я уже говорил, полный пример в основном связан с MapIterator и ExpectingIterator

Итераторы снова не становятся действительными на php < 7, ваш пример с arrayIterator и образец с MapIterator останавливаются после исчерпания первоначального пула...

С другой стороны, все это работает на более ранних версиях php, если вы используете → append method на итераторе вместо [] push.

Ответ 4

Ответ: да, вы можете. Вам просто нужно больше генераторов. И разделить логику запроса и логику очередей на асинхронный дизайн. Вместо того, чтобы использовать массив для запросов, которые ваш пул собирается выдавать, и ждать его, должен быть сам генератор, который дает новые запросы из вашего первоначального списка и запросы, добавленные из разобранных ответов, до тех пор, пока все запросы не будут отправлены разобранным, а результирующий запрос будет отправлен и встречается (повторяющееся) или условие остановки.