gmail-smtp / google-api-php-client / vendor / guzzlehttp / guzzle / src / Pool.php

Pool.php in Gmail SMTP trunk, at google-api-php-client/vendor/guzzlehttp/guzzle/src/Pool.php

138 lines 5.0 KB

Raw Download Zip

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