1
|
|
|
<?php |
2
|
|
|
|
3
|
|
|
namespace mpyw\Co; |
4
|
|
|
use mpyw\Co\Internal\Utils; |
5
|
|
|
use mpyw\Co\Internal\CoOption; |
6
|
|
|
use mpyw\Co\Internal\GeneratorContainer; |
7
|
|
|
use mpyw\Co\Internal\CURLPool; |
8
|
|
|
|
9
|
|
|
use React\Promise\Deferred; |
10
|
|
|
use React\Promise\PromiseInterface; |
11
|
|
|
use function React\Promise\all; |
12
|
|
|
|
13
|
|
|
class Co implements CoInterface |
14
|
|
|
{ |
15
|
|
|
/** |
16
|
|
|
* Instance of myself. |
17
|
|
|
* @var Co |
18
|
|
|
*/ |
19
|
|
|
private static $self; |
20
|
|
|
|
21
|
|
|
/** |
22
|
|
|
* Options. |
23
|
|
|
* @var CoOption |
24
|
|
|
*/ |
25
|
|
|
private $options; |
26
|
|
|
|
27
|
|
|
/** |
28
|
|
|
* cURL request pool object. |
29
|
|
|
* @var CURLPool |
30
|
|
|
*/ |
31
|
|
|
private $pool; |
32
|
|
|
|
33
|
|
|
/** |
34
|
|
|
* Overwrite CoOption default. |
35
|
|
|
* @param array $options |
36
|
|
|
*/ |
37
|
|
|
public static function setDefaultOptions(array $options) |
38
|
|
|
{ |
39
|
|
|
CoOption::setDefault($options); |
40
|
|
|
} |
41
|
|
|
|
42
|
|
|
/** |
43
|
|
|
* Get CoOption default as array. |
44
|
|
|
* @return array |
45
|
|
|
*/ |
46
|
|
|
public static function getDefaultOptions() |
47
|
|
|
{ |
48
|
|
|
return CoOption::getDefault(); |
49
|
|
|
} |
50
|
|
|
|
51
|
|
|
/** |
52
|
|
|
* Wait until value is recursively resolved to return it. |
53
|
|
|
* This function call must be atomic. |
54
|
|
|
* @param mixed $value |
55
|
|
|
* @param array $options |
56
|
|
|
* @return mixed |
57
|
|
|
*/ |
58
|
4 |
|
public static function wait($value, array $options = []) |
59
|
4 |
|
{ |
60
|
|
|
try { |
61
|
4 |
|
if (self::$self) { |
62
|
|
|
throw new \BadMethodCallException('Co::wait() is already running. Use Co::async() instead.'); |
63
|
|
|
} |
64
|
4 |
|
self::$self = new self; |
65
|
4 |
|
return self::$self->start($value, new CoOption($options)); |
66
|
|
|
} finally { |
67
|
4 |
|
self::$self = null; |
68
|
|
|
} |
69
|
|
|
} |
70
|
|
|
|
71
|
|
|
/** |
72
|
|
|
* Value is recursively resolved, but we never wait it. |
73
|
|
|
* This function must be called along with Co::wait(). |
74
|
|
|
* @param mixed $value |
75
|
|
|
* @param array $options |
76
|
|
|
*/ |
77
|
1 |
|
public static function async($value, array $options = []) |
78
|
1 |
|
{ |
79
|
1 |
|
if (!self::$self) { |
80
|
|
|
throw new \BadMethodCallException( |
81
|
|
|
'Co::async() must be called along with Co::wait(). ' . |
82
|
|
|
'This method is mainly expected to be used in CURLOPT_WRITEFUNCTION callback.' |
83
|
|
|
); |
84
|
|
|
} |
85
|
1 |
|
self::$self->start($value, self::$self->options->reconfigure($options), false); |
86
|
1 |
|
} |
87
|
|
|
|
88
|
|
|
/** |
89
|
|
|
* External instantiation is forbidden. |
90
|
|
|
*/ |
91
|
|
|
private function __construct() {} |
92
|
|
|
|
93
|
|
|
/** |
94
|
|
|
* Start resovling. |
95
|
|
|
* @param mixed $value |
96
|
|
|
* @param CoOption $options |
97
|
|
|
* @param bool $wait |
98
|
|
|
* @param mixed If $wait, return resolved value. |
99
|
|
|
*/ |
100
|
4 |
|
private function start($value, CoOption $options, $wait = true) |
101
|
4 |
|
{ |
102
|
4 |
|
$this->options = $options; |
103
|
4 |
|
$this->pool = new CURLPool($options); |
104
|
|
|
// If $wait, final result is stored into referenced $return |
105
|
4 |
|
if ($wait) { |
106
|
4 |
|
$deferred = new Deferred; |
107
|
|
|
$deferred->promise()->done(function ($r) use (&$return) { |
108
|
4 |
|
$return = $r; |
109
|
4 |
|
}); |
110
|
|
|
} |
111
|
|
|
// For convenience, all values are wrapped into generator |
112
|
|
|
$genfunc = function () use ($value) { |
113
|
4 |
|
yield CoInterface::RETURN_WITH => (yield $value); |
114
|
4 |
|
}; |
115
|
4 |
|
$con = Utils::normalize($genfunc, $options); |
116
|
|
|
// We have to provide deferred object only if $wait |
117
|
4 |
|
$this->processGeneratorContainer($con, $wait ? $deferred : null); |
118
|
|
|
// We have to wait $return only if $wait |
119
|
4 |
|
if ($wait) { |
120
|
4 |
|
$this->pool->wait(); |
121
|
4 |
|
return $return; |
122
|
|
|
} |
123
|
1 |
|
} |
124
|
|
|
|
125
|
|
|
/** |
126
|
|
|
* Handle resolving generators. |
127
|
|
|
* @param GeneratorContainer $gc |
128
|
|
|
* @param Deferred $deferred |
129
|
|
|
*/ |
130
|
4 |
|
private function processGeneratorContainer(GeneratorContainer $gc, Deferred $deferred = null) |
131
|
4 |
|
{ |
132
|
|
|
// If generator has no more yields... |
133
|
4 |
|
if (!$gc->valid()) { |
134
|
|
|
// If exception has been thrown in generator, we have to propagate it as rejected value |
135
|
4 |
|
if ($gc->thrown()) { |
136
|
|
|
$deferred && $deferred->reject($gc->getReturnOrThrown()); |
137
|
|
|
return; |
138
|
|
|
} |
139
|
|
|
// Now we normalize returned value |
140
|
|
|
try { |
141
|
4 |
|
$returned = Utils::normalize($gc->getReturnOrThrown(), $gc->getOptions()); |
142
|
4 |
|
$yieldables = Utils::getYieldables($returned); |
143
|
|
|
// If normalized value contains yieldables, we have to chain resolver |
144
|
4 |
|
if ($yieldables) { |
145
|
1 |
|
$this->promiseAll($yieldables, true)->then( |
146
|
1 |
|
self::getApplier($returned, $yieldables, [$deferred, 'resolve']), |
147
|
1 |
|
[$deferred, 'reject'] |
148
|
|
|
); |
149
|
1 |
|
return; |
150
|
|
|
} |
151
|
|
|
// Propagate normalized returned value |
152
|
4 |
|
$deferred && $deferred->resolve($returned); |
153
|
|
|
} catch (\RuntimeException $e) { |
154
|
|
|
// Propagate exception thrown in normalization |
155
|
|
|
$deferred && $deferred->reject($e); |
156
|
|
|
} |
157
|
4 |
|
return; |
158
|
|
|
} |
159
|
|
|
|
160
|
|
|
// Now we normalize yielded value |
161
|
|
|
try { |
162
|
4 |
|
$yielded = Utils::normalize($gc->current(), $gc->getOptions(), $gc->key()); |
163
|
|
|
} catch (\RuntimeException $e) { |
164
|
|
|
// If exception thrown in normalization... |
165
|
|
|
// - If generator accepts exception, we throw it into generator |
166
|
|
|
// - If generator does not accept exception, we assume it as non-exception value |
167
|
|
|
$gc->throwAcceptable() ? $gc->throw_($e) : $gc->send($e); |
168
|
|
|
// Continue |
169
|
|
|
$this->processGeneratorContainer($gc, $deferred); |
170
|
|
|
return; |
171
|
|
|
} |
172
|
|
|
|
173
|
|
|
// Search yieldables from yielded value |
174
|
4 |
|
$yieldables = Utils::getYieldables($yielded); |
175
|
4 |
|
if (!$yieldables) { |
176
|
|
|
// If there are no yieldables, send yielded value back into generator |
177
|
2 |
|
$gc->send($yielded); |
178
|
|
|
// Continue |
179
|
2 |
|
$this->processGeneratorContainer($gc, $deferred); |
180
|
2 |
|
return; |
181
|
|
|
} |
182
|
|
|
|
183
|
|
|
// Chain resolver |
184
|
3 |
|
$this->promiseAll($yieldables, $gc->throwAcceptable())->then( |
185
|
3 |
|
self::getApplier($yielded, $yieldables, [$gc, 'send']), |
186
|
3 |
|
[$gc, 'throw_'] |
187
|
|
|
)->always(function () use ($gc, $deferred) { |
188
|
|
|
// Continue |
189
|
3 |
|
$this->processGeneratorContainer($gc, $deferred); |
190
|
3 |
|
}); |
191
|
3 |
|
} |
192
|
|
|
|
193
|
|
|
/** |
194
|
|
|
* Return function that apply changes in yieldables. |
195
|
|
|
* @param mixed $yielded |
196
|
|
|
* @param array $yieldables |
197
|
|
|
* @param callable $next |
198
|
|
|
*/ |
199
|
3 |
|
private static function getApplier($yielded, $yieldables, callable $next) |
200
|
3 |
|
{ |
201
|
|
|
return function (array $results) use ($yielded, $yieldables, $next) { |
202
|
3 |
|
foreach ($results as $hash => $resolved) { |
203
|
3 |
|
$current = &$yielded; |
204
|
3 |
|
foreach ($yieldables[$hash]['keylist'] as $key) { |
205
|
3 |
|
$current = &$current[$key]; |
206
|
|
|
} |
207
|
3 |
|
$current = $resolved; |
208
|
|
|
} |
209
|
3 |
|
$next($yielded); |
210
|
3 |
|
}; |
211
|
|
|
} |
212
|
|
|
|
213
|
|
|
/** |
214
|
|
|
* Promise all changes in yieldables are prepared. |
215
|
|
|
* @param array $yieldables |
216
|
|
|
* @param bool $throw_acceptable |
217
|
|
|
* @return PromiseInterface |
218
|
|
|
*/ |
219
|
3 |
|
private function promiseAll($yieldables, $throw_acceptable) |
220
|
3 |
|
{ |
221
|
3 |
|
$promises = []; |
222
|
3 |
|
foreach ($yieldables as $yieldable) { |
223
|
3 |
|
$dfd = new Deferred; |
224
|
3 |
|
$promises[(string)$yieldable['value']] = $dfd->promise(); |
225
|
|
|
// If caller cannot accept exception, |
226
|
|
|
// we handle rejected value as resolved. |
227
|
3 |
|
if (!$throw_acceptable) { |
228
|
1 |
|
$original_dfd = $dfd; |
229
|
1 |
|
$dfd = new Deferred; |
230
|
1 |
|
$absorber = function ($any) use ($original_dfd) { |
231
|
1 |
|
$original_dfd->resolve($any); |
232
|
1 |
|
}; |
233
|
1 |
|
$dfd->promise()->then($absorber, $absorber); |
234
|
|
|
} |
235
|
|
|
// Add or enqueue cURL handles |
236
|
3 |
|
if (Utils::isCurl($yieldable['value'])) { |
237
|
2 |
|
$this->pool->addOrEnqueue($yieldable['value'], $dfd); |
238
|
2 |
|
continue; |
239
|
|
|
} |
240
|
|
|
// Process generators |
241
|
2 |
|
if (Utils::isGeneratorContainer($yieldable['value'])) { |
242
|
2 |
|
$this->processGeneratorContainer($yieldable['value'], $dfd); |
243
|
2 |
|
continue; |
244
|
|
|
} |
245
|
|
|
} |
246
|
3 |
|
return all($promises); |
247
|
|
|
} |
248
|
|
|
} |
249
|
|
|
|