media-cloud-sync / includes / sdk / s3 / GuzzleHttp / Pool.php

Pool.php in Media Cloud Sync 1.2.6, at includes/sdk/s3/GuzzleHttp/Pool.php

118 lines 4.7 KB

Raw Download Zip

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