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