stockpack / vendor / guzzlehttp / guzzle / src / Pool.php

Pool.php in StockPack – Stock photos and AI images from Unsplash, Adobe Stock, Freepik and more 2.4.3, at vendor/guzzlehttp/guzzle/src/Pool.php

133 lines 4.7 KB

Raw Download Zip

1	<?php
2	namespace GuzzleHttp;
3
4	use GuzzleHttp\Promise\EachPromise;
5	use GuzzleHttp\Promise\PromisorInterface;
6	use Psr\Http\Message\RequestInterface;
7
8	/**
9	* Sends an iterator of requests concurrently using a capped pool size.
10	*
11	* The pool will read from an iterator until it is cancelled or until the
12	* iterator is consumed. When a request is yielded, the request is sent after
13	* applying the "request_options" request options (if provided in the ctor).
14	*
15	* When a function is yielded by the iterator, the function is provided the
16	* "request_options" array that should be merged on top of any existing
17	* options, and the function MUST then return a wait-able promise.
18	*/
19	class Pool implements PromisorInterface
20	{
21	/** @var EachPromise */
22	private $each;
23
24	/**
25	* @param ClientInterface $client Client used to send the requests.
26	* @param array\|\Iterator $requests Requests or functions that return
27	* requests to send concurrently.
28	* @param array $config Associative array of options
29	* - concurrency: (int) Maximum number of requests to send concurrently
30	* - options: Array of request options to apply to each request.
31	* - fulfilled: (callable) Function to invoke when a request completes.
32	* - rejected: (callable) Function to invoke when a request is rejected.
33	*/
34	public function __construct(
35	ClientInterface $client,
36	$requests,
37	array $config = []
38	) {
39	// Backwards compatibility.
40	if (isset($config['pool_size'])) {
41	$config['concurrency'] = $config['pool_size'];
42	} elseif (!isset($config['concurrency'])) {
43	$config['concurrency'] = 25;
44	}
45
46	if (isset($config['options'])) {
47	$opts = $config['options'];
48	unset($config['options']);
49	} else {
50	$opts = [];
51	}
52
53	$iterable = \GuzzleHttp\Promise\iter_for($requests);
54	$requests = function () use ($iterable, $client, $opts) {
55	foreach ($iterable as $key => $rfn) {
56	if ($rfn instanceof RequestInterface) {
57	yield $key => $client->sendAsync($rfn, $opts);
58	} elseif (is_callable($rfn)) {
59	yield $key => $rfn($opts);
60	} else {
61	throw new \InvalidArgumentException('Each value yielded by '
62	. 'the iterator must be a Psr7\Http\Message\RequestInterface '
63	. 'or a callable that returns a promise that fulfills '
64	. 'with a Psr7\Message\Http\ResponseInterface object.');
65	}
66	}
67	};
68
69	$this->each = new EachPromise($requests(), $config);
70	}
71
72	/**
73	* Get promise
74	* @return GuzzleHttp\Promise\Promise
75	*/
76	public function promise()
77	{
78	return $this->each->promise();
79	}
80
81	/**
82	* Sends multiple requests concurrently and returns an array of responses
83	* and exceptions that uses the same ordering as the provided requests.
84	*
85	* IMPORTANT: This method keeps every request and response in memory, and
86	* as such, is NOT recommended when sending a large number or an
87	* indeterminate number of requests concurrently.
88	*
89	* @param ClientInterface $client Client used to send the requests
90	* @param array\|\Iterator $requests Requests to send concurrently.
91	* @param array $options Passes through the options available in
92	* {@see GuzzleHttp\Pool::__construct}
93	*
94	* @return array Returns an array containing the response or an exception
95	* in the same order that the requests were sent.
96	* @throws \InvalidArgumentException if the event format is incorrect.
97	*/
98	public static function batch(
99	ClientInterface $client,
100	$requests,
101	array $options = []
102	) {
103	$res = [];
104	self::cmpCallback($options, 'fulfilled', $res);
105	self::cmpCallback($options, 'rejected', $res);
106	$pool = new static($client, $requests, $options);
107	$pool->promise()->wait();
108	ksort($res);
109
110	return $res;
111	}
112
113	/**
114	* Execute callback(s)
115	*
116	* @return void
117	*/
118	private static function cmpCallback(array &$options, $name, array &$results)
119	{
120	if (!isset($options[$name])) {
121	$options[$name] = function ($v, $k) use (&$results) {
122	$results[$k] = $v;
123	};
124	} else {
125	$currentFn = $options[$name];
126	$options[$name] = function ($v, $k) use (&$results, $currentFn) {
127	$currentFn($v, $k);
128	$results[$k] = $v;
129	};
130	}
131	}
132	}
133