[10.x] Feature: Enhanced Request Pooling with Concurrency Limit Control

src/Illuminate/Http/Client/PendingRequest.php

@@ -11,6 +11,7 @@
 use GuzzleHttp\Exception\TransferException;
 use GuzzleHttp\HandlerStack;
 use GuzzleHttp\Middleware;
+use GuzzleHttp\Promise\Each;
 use GuzzleHttp\UriTemplate\UriTemplate;
 use Illuminate\Contracts\Support\Arrayable;
 use Illuminate\Http\Client\Events\ConnectionFailed;
@@ -846,17 +847,31 @@ public function delete(string $url, $data = [])
      * Send a pool of asynchronous requests concurrently.
      *
      * @param  callable  $callback
+     * @param  int|null  $concurrencyLimit
      * @return array<array-key, \Illuminate\Http\Client\Response>
      */
-    public function pool(callable $callback)
+    public function pool(callable $callback, ?int $concurrencyLimit = null)
     {
         $results = [];
 
-        $requests = tap(new Pool($this->factory), $callback)->getRequests();
-
-        foreach ($requests as $key => $item) {
-            $results[$key] = $item instanceof static ? $item->getPromise()->wait() : $item->wait();
-        }
+        $pool = tap(new Pool($this->factory), $callback)->getRequests();
+
+        $promises = array_map(function ($request) {
+            return $request instanceof static ? $request->getPromise() : $request;
+        }, $pool);
+
+        Each::ofLimit(
+            $promises,
+            $concurrencyLimit ?: count($pool),
+            function ($response, $index) use (&$results) {
+                $results[$index] = $response instanceof \GuzzleHttp\Psr7\Response
+                    ? new \Illuminate\Http\Client\Response($response)
+                    : $response;
+            },
+            function ($reason, $index) use (&$results) {
+                $results[$index] = $reason;
+            }
+        )->wait();
 
         return $results;
     }

tests/Http/HttpClientTest.php

@@ -1540,6 +1540,38 @@ public function testMultipleRequestsAreSentInThePoolWithKeys()
         $this->assertSame(500, $responses['test500']->status());
     }
 
+    public function testRequestConcurrencyLimitInPoolIsRespected()
+    {
+        $this->factory->fake([
+            'https://laravel.com/*' => function ($request) {
+                return $this->factory::response('', 200);
+            },
+        ]);
+
+        $callback = function (Pool $pool) {
+            for ($i = 0; $i < 10; $i++) {
+                $pool->as("request-{$i}")->async()->get("https://laravel.com/test-{$i}");
+            }
+        };
+
+        $concurrencyLimits = [2, 10];
+        $previousDuration = PHP_INT_MAX;
+
+        foreach ($concurrencyLimits as $limit) {
+            $startTime = microtime(true);
+
+            $results = $this->factory->pool($callback, $limit);
+
+            $endTime = microtime(true);
+            $duration = $endTime - $startTime;
+
+            $this->assertCount(10, $results);
+            $this->assertLessThanOrEqual($previousDuration, $duration);
+
+            $previousDuration = $duration;
+        }
+    }
+
     public function testMiddlewareRunsInPool()
     {
         $this->factory->fake(function (Request $request) {