在 Guzzle 中处理 EachPromise (each_limit) 的聚合承诺

Question

任务

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

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

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

工具集

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

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

• fulfilled：（可调用的）在 promise 完成时调用。该函数使用三个参数调用：实现值、索引
  来自承诺的可迭代列表的位置，以及聚合
  管理所有承诺的承诺。总承诺可能
  从回调中解决以短路承诺。
• rejected：（可调用）当一个承诺被拒绝时调用。该函数使用三个参数调用：拒绝原因、
  来自承诺的可迭代列表的索引位置，以及
  管理所有承诺的聚合承诺。聚合
  承诺可以从回调中解决到短路
  承诺。
• 并发：（整数）传递此配置选项以限制允许的未完成并发执行的承诺数，
  创建一个有上限的承诺池。默认没有限制。

代码

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

问题

在配置哈希中，fulfilled 回调接收 3 个对象，如文档所述。 $promiseResult 可以正确处理，$paginatedResult->getItems() 实际上返回请求页面中的项目数组，但我无法聚合这些项目。 $aggregatedResult 是 null，最后一个 thened 完成回调中的 $promisedAggregatedResult 也是 null。

问题

如何正确使用 Guzzle 的 EachPromise（以及它的辅助函数 each 和 each_limit）来汇总传递给它的所有承诺的结果？

Answer 1

您也可以使用Pool::batch() 方法，它在内部几乎相同。

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

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

Answer 2

原因

根据 EachPromise 类描述：

表示一个遍历多个 Promise 并调用的 Promise 过程中的副作用函数。

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

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

实际代码解决方案

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