media-cloud-sync / includes / sdk / s3 / Aws / CommandPool.php

CommandPool.php in Media Cloud Sync 1.2.12, at includes/sdk/s3/Aws/CommandPool.php

132 lines 5.2 KB

Raw Download Zip

1	<?php
2
3	namespace Dudlewebs\WPMCS\s3\Aws;
4
5	use Dudlewebs\WPMCS\s3\GuzzleHttp\Promise\PromiseInterface;
6	use Dudlewebs\WPMCS\s3\GuzzleHttp\Promise\PromisorInterface;
7	use Dudlewebs\WPMCS\s3\GuzzleHttp\Promise\EachPromise;
8	/**
9	* Sends and iterator of commands concurrently using a capped pool size.
10	*
11	* The pool will read command objects from an iterator until it is cancelled or
12	* until the iterator is consumed.
13	*/
14	class CommandPool implements PromisorInterface
15	{
16	/** @var EachPromise */
17	private $each;
18	/**
19	* The CommandPool constructor accepts a hash of configuration options:
20	*
21	* - concurrency: (callable\|int) Maximum number of commands to execute
22	* concurrently. Provide a function to resize the pool dynamically. The
23	* function will be provided the current number of pending requests and
24	* is expected to return an integer representing the new pool size limit.
25	* - before: (callable) function to invoke before sending each command. The
26	* before function accepts the command and the key of the iterator of the
27	* command. You can mutate the command as needed in the before function
28	* before sending the command.
29	* - fulfilled: (callable) Function to invoke when a promise is fulfilled.
30	* The function is provided the result object, id of the iterator that the
31	* result came from, and the aggregate promise that can be resolved/rejected
32	* if you need to short-circuit the pool.
33	* - rejected: (callable) Function to invoke when a promise is rejected.
34	* The function is provided an AwsException object, id of the iterator that
35	* the exception came from, and the aggregate promise that can be
36	* resolved/rejected if you need to short-circuit the pool.
37	* - preserve_iterator_keys: (bool) Retain the iterator key when generating
38	* the commands.
39	*
40	* @param AwsClientInterface $client Client used to execute commands.
41	* @param array\|\Iterator $commands Iterable that yields commands.
42	* @param array $config Associative array of options.
43	*/
44	public function __construct(AwsClientInterface $client, $commands, array $config = [])
45	{
46	if (!isset($config['concurrency'])) {
47	$config['concurrency'] = 25;
48	}
49	$before = $this->getBefore($config);
50	$mapFn = function ($commands) use($client, $before, $config) {
51	foreach ($commands as $key => $command) {
52	if (!$command instanceof CommandInterface) {
53	throw new \InvalidArgumentException('Each value yielded by ' . 'the iterator must be an Aws\\CommandInterface.');
54	}
55	if ($before) {
56	$before($command, $key);
57	}
58	if (!empty($config['preserve_iterator_keys'])) {
59	(yield $key => $client->executeAsync($command));
60	} else {
61	(yield $client->executeAsync($command));
62	}
63	}
64	};
65	$this->each = new EachPromise($mapFn($commands), $config);
66	}
67	/**
68	* @return PromiseInterface
69	*/
70	public function promise()
71	{
72	return $this->each->promise();
73	}
74	/**
75	* Executes a pool synchronously and aggregates the results of the pool
76	* into an indexed array in the same order as the passed in array.
77	*
78	* @param AwsClientInterface $client Client used to execute commands.
79	* @param mixed $commands Iterable that yields commands.
80	* @param array $config Configuration options.
81	*
82	* @return array
83	* @see \Aws\CommandPool::__construct for available configuration options.
84	*/
85	public static function batch(AwsClientInterface $client, $commands, array $config = [])
86	{
87	$results = [];
88	self::cmpCallback($config, 'fulfilled', $results);
89	self::cmpCallback($config, 'rejected', $results);
90	return (new self($client, $commands, $config))->promise()->then(static function () use(&$results) {
91	\ksort($results);
92	return $results;
93	})->wait();
94	}
95	/**
96	* @return callable
97	*/
98	private function getBefore(array $config)
99	{
100	if (!isset($config['before'])) {
101	return null;
102	}
103	if (\is_callable($config['before'])) {
104	return $config['before'];
105	}
106	throw new \InvalidArgumentException('before must be callable');
107	}
108	/**
109	* Adds an onFulfilled or onRejected callback that aggregates results into
110	* an array. If a callback is already present, it is replaced with the
111	* composed function.
112	*
113	* @param array $config
114	* @param $name
115	* @param array $results
116	*/
117	private static function cmpCallback(array &$config, $name, array &$results)
118	{
119	if (!isset($config[$name])) {
120	$config[$name] = function ($v, $k) use(&$results) {
121	$results[$k] = $v;
122	};
123	} else {
124	$currentFn = $config[$name];
125	$config[$name] = function ($v, $k) use(&$results, $currentFn) {
126	$currentFn($v, $k);
127	$results[$k] = $v;
128	};
129	}
130	}
131	}
132