Completed
Push — master ( 097a75...93608a )
by Ryosuke
03:30
created

Co::__construct()   A

Complexity

Conditions 1
Paths 1

Size

Total Lines 1
Code Lines 1

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 2

Importance

Changes 4
Bugs 0 Features 0
Metric Value
c 4
b 0
f 0
dl 0
loc 1
ccs 0
cts 0
cp 0
rs 10
cc 1
eloc 1
nc 1
nop 0
crap 2
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