在大炮中处理每个promise（each_limit）的总promise

Question

假设我们有一个API POSTendpoint，它返回如下答案：

{
  "data": [1,2,3],
  "total_rows": 20
}

这意味着我们收到了部分数据，只有3个条目的第一页。总条目数为20，这意味着我们希望在请求中使用一些偏移量调用其他页面。因此，总计数（以及可能的偏移量）只有在第一次调用完成后才知道，而其他调用并不相互依赖，可以同时完成。

它是PHP，对于这个任务，Guzzle6与Promises/A一起使用。此外，Guzzes提供了EachPromise类，该类接收多个promise的迭代和用于设置处理的配置哈希。根据类构造函数的PhpDoc块：

配置哈希可以包括以下键值对：

• 已履行：（可调用）在promise履行时调用。调用该函数时有三个参数：履行值、promise的iterable列表中的索引位置，以及管理所有promise的聚合promise。聚合promise可以
  从回调内部解决，以缩短promise

$paginatedResult = $this->client->sendAsync($this->createRequest($requestBody))
    ->then(function (ResponseInterface $response) use ($requestBody) {
            return $this->deserializeToPaginatedResult(
                $response,
                $requestBody->paginatedResultClass()
            );
        }
    )->wait();

$pageGenerator = function () use ($paginatedResult, $requestBody) {
    $perPageCount = count($paginatedResult->getItems());
    $totalItems = $paginatedResult->getTotalCount();

    for ($currentOffset = $perPageCount; $currentOffset <= $totalItems; $currentOffset += $perPageCount) {
        $newRequestBody = clone $requestBody;
        $newRequestBody->setOffset($currentOffset);

        yield $this->client->sendAsync($this->createRequest($newRequestBody));
    }
};

$aggregatedResult = (new EachPromise(
    $pageGenerator(), [
        'concurrency' => 4,
        'fulfilled' => function ($promiseResult, $promiseIndex, $promiseAggregate) use ($requestBody) {
            $paginatedResult = $this->deserializeToPaginatedResult(
                $promiseResult,
                $requestBody->paginatedResultClass()
            );

            return $paginatedResult->getItems();
        },
    ]
))->promise()
    ->then(
        function ($promisedAggregatedResult) {
            var_dump($promisedAggregatedResult);
        }
    )
    ->wait();

var_dump($aggregatedResult);

在配置散列中，completed回调接收3个对象，如文档所述。可以正确处理$promiseResult，并且$paginatedResult-

应该如何正确地使用古斯的每个promise（以及它的助手函数）来聚合传递给它的所有promise的结果？




Answer 1

匿名用户

您也可以使用池::batch（）方法，它在内部执行几乎相同的操作。

$pageGenerator = function () use (...) {
    // ...
    yield $this->createRequest($newRequestBody);
    // ...
}

$responses = Pool::batch($this->client, $pageGenerator(), ['concurrency' => 4]);

Answer 2

根据各Promise类别说明：

表示对多个promise进行迭代并在流程中调用副作用函数的promise。

因此，这些配置哈希可调用函数只是副作用函数，可能会或可能不会短路解决聚合promise。这正是实现函数既不影响$promiseggregate，也不以最后一个可实现调用的参数$promisedAggregatedResult结束的原因。

关于正确用法的提示可以在\GuzzleHttp\Promise\all和\GuzzleHttp\Promise\some函数中找到。关键的想法是在那些副作用实现的回调中通过引用使用外部聚合。如果聚合promise从未在这些副作用函数中解析，那么它将解析为null，然后在下一个然后履行回调中传递。同样，应该在可调用的实现中使用通过引用聚合，以便将其作为promise实现值返回。

$paginatedResult = $this->client->sendAsync($this->createRequest($requestBody))
    ->then(function (ResponseInterface $response) use ($requestBody) {
            return $this->deserializeToPaginatedResult(
                $response,
                $requestBody->paginatedResultClass()
            );
        }
    )->wait();

$pageGenerator = function () use ($paginatedResult, $requestBody) {
    $perPageCount = count($paginatedResult->getItems());
    $totalItems = $paginatedResult->getTotalCount();

    for ($currentOffset = $perPageCount; $currentOffset <= $totalItems; $currentOffset += $perPageCount) {
        $newRequestBody = clone $requestBody;
        $newRequestBody->setOffset($currentOffset);

        yield $this->client->sendAsync($this->createRequest($newRequestBody));
    }
};
$items = $paginatedResult->getItems();

return each_limit(
    $pageGenerator(),
    4,
    function ($promiseResult) use (&$items, $requestBody) {
        $paginatedResult = $this->deserializeToPaginatedResult(
            $promiseResult,
            $requestBody->paginatedResultClass()
        );
        $items = array_merge($items, $paginatedResult->getItems());
    },
    function ($reason, $idx, PromiseInterface $aggregate) {
        $aggregate->reject($reason);
    })->then(function () use (&$items) {
        ksort($items);
        return $items;
    });