[12.x] Add concurrency control to Http::pool and Http::batch (#57555)

* Added concurrency to Http::pool

* Added concurrency to Http::batch

* formatting

---------

Co-authored-by: Taylor Otwell <taylor@laravel.com>

Author: WendellAdriel

src/Illuminate/Http/Client/Batch.php

@@ -114,6 +114,13 @@ class Batch
      */
     public $finishedAt = null;
 
+    /**
+     * The maximum number of concurrent requests.
+     *
+     * @var int|null
+     */
+    protected $concurrencyLimit = null;
+
     /**
      * Create a new request batch instance.
      */
@@ -208,6 +215,19 @@ public function finally(Closure $callback): self
         return $this;
     }
 
+    /**
+     * Set the maximum number of concurrent requests.
+     *
+     * @param  int  $limit
+     * @return Batch
+     */
+    public function concurrency(int $limit): self
+    {
+        $this->concurrencyLimit = $limit;
+
+        return $this;
+    }
+
     /**
      * Defer the batch to run in the background after the current task has finished.
      *
@@ -244,7 +264,7 @@ public function send(): array
         }
 
         if (! empty($promises)) {
-            (new EachPromise($promises, [
+            $eachPromiseOptions = [
                 'fulfilled' => function ($result, $key) use (&$results) {
                     $results[$key] = $result;
 
@@ -285,7 +305,13 @@ public function send(): array
 
                     return $reason;
                 },
-            ]))->promise()->wait();
+            ];
+
+            if ($this->concurrencyLimit !== null) {
+                $eachPromiseOptions['concurrency'] = $this->concurrencyLimit;
+            }
+
+            (new EachPromise($promises, $eachPromiseOptions))->promise()->wait();
         }
 
         // Before returning the results, we must ensure that the results are sorted

src/Illuminate/Http/Client/PendingRequest.php

@@ -11,6 +11,7 @@
 use GuzzleHttp\Exception\TransferException;
 use GuzzleHttp\HandlerStack;
 use GuzzleHttp\Middleware;
+use GuzzleHttp\Promise\EachPromise;
 use GuzzleHttp\UriTemplate\UriTemplate;
 use Illuminate\Contracts\Support\Arrayable;
 use Illuminate\Http\Client\Events\ConnectionFailed;
@@ -884,18 +885,39 @@ public function delete(string $url, $data = [])
      * Send a pool of asynchronous requests concurrently.
      *
      * @param  callable  $callback
+     * @param  int|null  $concurrency
      * @return array<array-key, \Illuminate\Http\Client\Response>
      */
-    public function pool(callable $callback)
+    public function pool(callable $callback, ?int $concurrency = null)
     {
         $results = [];
 
         $requests = tap(new Pool($this->factory), $callback)->getRequests();
 
+        if ($concurrency === null) {
+            foreach ($requests as $key => $item) {
+                $results[$key] = $item instanceof static ? $item->getPromise()->wait() : $item->wait();
+            }
+
+            return $results;
+        }
+
+        $promises = [];
+
         foreach ($requests as $key => $item) {
-            $results[$key] = $item instanceof static ? $item->getPromise()->wait() : $item->wait();
+            $promises[$key] = $item instanceof static ? $item->getPromise() : $item;
         }
 
+        (new EachPromise($promises, [
+            'fulfilled' => function ($result, $key) use (&$results) {
+                $results[$key] = $result;
+            },
+            'rejected' => function ($reason, $key) use (&$results) {
+                $results[$key] = $reason;
+            },
+            'concurrency' => $concurrency,
+        ]))->promise()->wait();
+
         return $results;
     }
 

tests/Http/HttpClientTest.php

@@ -1910,6 +1910,27 @@ public function testMiddlewareRunsInPool()
         $this->assertSame(['hyped-for' => 'laravel-movie'], json_decode(tap($history[0]['request']->getBody())->rewind()->getContents(), true));
     }
 
+    public function testPoolConcurrency()
+    {
+        $this->factory->fake([
+            '200.com' => $this->factory::response('', 200),
+            '400.com' => $this->factory::response('', 400),
+            '500.com' => $this->factory::response('', 500),
+        ]);
+
+        $responses = $this->factory->pool(function (Pool $pool) {
+            return [
+                $pool->get('200.com'),
+                $pool->get('400.com'),
+                $pool->get('500.com'),
+            ];
+        }, 2);
+
+        $this->assertSame(200, $responses[0]->status());
+        $this->assertSame(400, $responses[1]->status());
+        $this->assertSame(500, $responses[2]->status());
+    }
+
     public function testTheRequestSendingAndResponseReceivedEventsAreFiredWhenARequestIsSent()
     {
         $events = m::mock(Dispatcher::class);
@@ -4068,6 +4089,42 @@ public function testBatchFinallyHookIsCalledWithErrors(): void
         $this->assertSame($responses['second'], $finallyCallback['second']);
     }
 
+    public function testBatchConcurrency(): void
+    {
+        $this->factory->fake([
+            'https://200.com' => $this->factory::response('OK', 200),
+            'https://201.com' => $this->factory::response('Created', 201),
+            'https://202.com' => $this->factory::response('Accepted', 202),
+            'https://203.com' => $this->factory::response('Non-Authoritative Information', 203),
+        ]);
+
+        $executionOrder = [];
+
+        $batch = $this->factory->batch(function (Batch $batch) {
+            return [
+                $batch->as('first')->get('https://200.com'),
+                $batch->as('second')->get('https://201.com'),
+                $batch->as('third')->get('https://202.com'),
+                $batch->as('fourth')->get('https://203.com'),
+            ];
+        })->progress(function (Batch $batch, int|string $key, Response $response) use (&$executionOrder) {
+            $executionOrder[] = $key;
+        })->concurrency(2);
+
+        $this->assertSame(4, $batch->totalRequests);
+
+        $responses = $batch->send();
+
+        $this->assertSame(200, $responses['first']->status());
+        $this->assertSame(201, $responses['second']->status());
+        $this->assertSame(202, $responses['third']->status());
+        $this->assertSame(203, $responses['fourth']->status());
+
+        $this->assertSame(4, $batch->totalRequests);
+        $this->assertSame(0, $batch->pendingRequests);
+        $this->assertSame(0, $batch->failedRequests);
+    }
+
     public static function methodsReceivingArrayableDataProvider()
     {
         return [