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