Co::processGeneratorContainerRunning() - Code Metrics - mpyw/co - Measure and Improve Code Quality continuously with Scrutinizer

Co::processGeneratorContainerRunning() A
last analyzed 2019-12-10 22:57 UTC

↳ Parent: Co

Complexity

Conditions	3
Paths	3

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	21
CRAP Score	3

Importance

Changes

Metric	Value
dl	0
loc	32
ccs	21
cts	21
cp	1
rs	9.408
c	0
b	0
f	0
cc	3
nc	3
nop	1
crap	3

<?php

namespace mpyw\Co;
use mpyw\Co\Internal\TypeUtils;
use mpyw\Co\Internal\ControlUtils;
use mpyw\Co\Internal\YieldableUtils;
use mpyw\Co\Internal\CoOption;
use mpyw\Co\Internal\GeneratorContainer;
use mpyw\Co\Internal\Pool;
use React\Promise\Deferred;
use React\Promise\PromiseInterface;
use React\Promise\ExtendedPromiseInterface;
use React\Promise\FulfilledPromise;
use React\Promise\RejectedPromise;

class Co implements CoInterface
{
    /**
     * Instance of myself.
     * @var Co
     */
    private static $self;

    /**
     * Options.
     * @var CoOption
     */
    private $options;

    /**
     * cURL request pool object.
     * @var Pool
     */
    private $pool;

    /**
     * Running cURL or Generator identifiers.
     * @var array
     */
    private $runners = [];

    /**
     * Overwrite CoOption default.
     * @param array $options
     */
    public static function setDefaultOptions(array $options)
    {
        CoOption::setDefault($options);
    }

    /**
     * Get CoOption default as array.
     * @return array
     */
    public static function getDefaultOptions()
    {
        return CoOption::getDefault();
    }

    /**
     * Wait until value is recursively resolved to return it.
     * This function call must be atomic.
     * @param  mixed $value
     * @param  array $options
     * @return mixed
     */
    public static function wait($value, array $options = [])
    {
        try {
            if (self::$self) {
                throw new \BadMethodCallException('Co::wait() is already running. Use Co::async() instead.');
            }
            self::$self = new self;
            self::$self->options = new CoOption($options);
            self::$self->pool = new Pool(self::$self->options);
            return self::$self->start($value);
        } finally {
            self::$self = null;
        }
        // @codeCoverageIgnoreStart
    }
    // @codeCoverageIgnoreEnd

    /**
     * Value is recursively resolved, but we never wait it.
     * This function must be called along with Co::wait().
     * @param  mixed $value
     * @param  mixed $throw
     */
    public static function async($value, $throw = null)
    {
        if (!self::$self) {
            throw new \BadMethodCallException('Co::async() must be called along with Co::wait(). ');
        }
        if ($throw !== null) {
            $throw = filter_var($throw, FILTER_VALIDATE_BOOLEAN, [
                'flags' => FILTER_NULL_ON_FAILURE,
            ]);
            if ($throw === null) {
                throw new \InvalidArgumentException("\$throw must be null or boolean.");
            }
        }
        self::$self->start($value, false, $throw);
    }

    /**
     * Return if Co::wait() is running.
     * @return bool
     */
    public static function isRunning()
    {
        return (bool)self::$self;
    }

    /**
     * External instantiation is forbidden.
     */
    private function __construct() {}

    /**
     * Start resovling.
     * @param  mixed    $value
     * @param  bool     $wait
     * @param  mixed    $throw  Used for Co::async() overrides.
     * @param  mixed    If $wait, return resolved value.
     */
    private function start($value, $wait = true, $throw = null)
    {
        $return = null;
        // For convenience, all values are wrapped into generator
        $con = YieldableUtils::normalize($this->getRootGenerator($throw, $value, $return));
        $promise = $this->processGeneratorContainerRunning($con);
        if ($promise instanceof ExtendedPromiseInterface) {
            // This is actually 100% true; just used for unwrapping Exception thrown.
            $promise->done();
        }
        // We have to wait $return only if $wait
        if ($wait) {
            $this->pool->wait();
            return $return;
        }
    }

    /**
     * Handle resolving generators.
     * @param  GeneratorContainer $gc
     * @return PromiseInterface
     */
    private function processGeneratorContainer(GeneratorContainer $gc)
    {
        return $gc->valid()
            ? $this->processGeneratorContainerRunning($gc)
            : $this->processGeneratorContainerDone($gc);
    }

    /**
     * Handle resolving generators already done.
     * @param  GeneratorContainer $gc
     * @return PromiseInterface
     */
    private function processGeneratorContainerDone(GeneratorContainer $gc)
    {
        // If exception has been thrown in generator, we have to propagate it as rejected value
        if ($gc->thrown()) {
            return new RejectedPromise($gc->getReturnOrThrown());
        }

        // Now we normalize returned value
        $returned = YieldableUtils::normalize($gc->getReturnOrThrown(), $gc->getYieldKey());
        $yieldables = YieldableUtils::getYieldables($returned, [], $this->runners);

        // If normalized value contains yieldables, we have to chain resolver
        if ($yieldables) {
            $deferred = new Deferred;
            return $this->promiseAll($yieldables, true)
            ->then(
                YieldableUtils::getApplier($returned, $yieldables, [$deferred, 'resolve']),
                [$deferred, 'reject']
            )
            ->then(function () use ($yieldables, $deferred) {
                $this->runners = array_diff_key($this->runners, $yieldables);
                return $deferred->promise();
            });
        }

        // Propagate normalized returned value
        return new FulfilledPromise($returned);
    }

    /**
     * Handle resolving generators still running.
     * @param  GeneratorContainer $gc
     * @return PromiseInterface
     */
    private function processGeneratorContainerRunning(GeneratorContainer $gc)
    {
        // Check delay request yields
        if ($gc->key() === CoInterface::DELAY) {
            return $this->pool->addDelay($gc->current())
            ->then(function () use ($gc) {
                $gc->send(null);
                return $this->processGeneratorContainer($gc);
            });
        }

        // Now we normalize yielded value
        $yielded = YieldableUtils::normalize($gc->current());
        $yieldables = YieldableUtils::getYieldables($yielded, [], $this->runners);
        if (!$yieldables) {
            // If there are no yieldables, send yielded value back into generator
            $gc->send($yielded);
            // Continue
            return $this->processGeneratorContainer($gc);
        }

        // Chain resolver
        return $this->promiseAll($yieldables, $gc->key() !== CoInterface::SAFE)
        ->then(
            YieldableUtils::getApplier($yielded, $yieldables, [$gc, 'send']),
            [$gc, 'throw_']
        )->then(function () use ($gc, $yieldables) {
            // Continue
            $this->runners = array_diff_key($this->runners, $yieldables);
            return $this->processGeneratorContainer($gc);
        });
    }

    /**
     * Return root wrapper generator.
     * @param  mixed  $throw
     * @param  mixed  $value
     * @param  mixed  &$return
     */
    private function getRootGenerator($throw, $value, &$return)
    {
        try {
            if ($throw !== null) {
                $key = $throw ? null : CoInterface::SAFE;
            } else {
                $key = $this->options['throw'] ? null : CoInterface::SAFE;
            }
            $return = (yield $key => $value);
            return;
        } catch (\Throwable $e) {} catch (\Exception $e) {}
        $this->pool->reserveHaltException($e);
    }

    /**
     * Promise all changes in yieldables are prepared.
     * @param  array $yieldables
     * @param  bool  $throw_acceptable
     * @return PromiseInterface
     */
    private function promiseAll(array $yieldables, $throw_acceptable)
    {
        $promises = [];
        foreach ($yieldables as $yieldable) {
            // Add or enqueue cURL handles
            if (TypeUtils::isCurl($yieldable['value'])) {
                $promises[(string)$yieldable['value']] = $this->pool->addCurl($yieldable['value']);
                continue;
            }
            // Process generators
            if (TypeUtils::isGeneratorContainer($yieldable['value'])) {
                $promises[(string)$yieldable['value']] = $this->processGeneratorContainer($yieldable['value']);
                continue;
            }
        }
        // If caller cannot accept exception,
        // we handle rejected value as resolved.
        if (!$throw_acceptable) {
            $promises = array_map(
                ['\mpyw\Co\Internal\YieldableUtils', 'safePromise'],
                $promises
            );
        }
        return \React\Promise\all($promises);
    }

    /**
     * Wrap value with the Generator that returns the first successful result.
     * If all yieldables failed, AllFailedException is thrown.
     * If no yieldables found, AllFailedException is thrown.
     *
     * @param  mixed $value
     * @return \Generator Resolved value.
     * @throws AllFailedException
     */
    public static function any($value)
    {
        yield Co::RETURN_WITH => (yield ControlUtils::anyOrRace(
            $value,
            ['\mpyw\Co\Internal\ControlUtils', 'reverse'],
            'Co::any() failed.'
        ));
        // @codeCoverageIgnoreStart
    }
    // @codeCoverageIgnoreEnd

    /**
     * Wrap value with the Generator that returns the first result.
     * If no yieldables found, AllFailedException is thrown.
     *
     * @param  mixed $value
     * @return \Generator Resolved value.
     * @throws \RuntimeException|AllFailedException
     */
    public static function race($value)
    {
        yield Co::RETURN_WITH => (yield ControlUtils::anyOrRace(
            $value,
            ['\mpyw\Co\Internal\ControlUtils', 'fail'],
            'Co::race() failed.'
        ));
        // @codeCoverageIgnoreStart
    }
    // @codeCoverageIgnoreEnd

    /**
     * Wrap value with the Generator that returns the all results.
     * Normally you don't have to use this method, just yield an array that contains yieldables.
     * You should use only with Co::race() or Co::any().
     *
     * @param  mixed $value
     * @return \Generator Resolved value.
     * @throws \RuntimeException
     */
    public static function all($value)
    {
        yield Co::RETURN_WITH => (yield $value);
        // @codeCoverageIgnoreStart
    }
    // @codeCoverageIgnoreEnd
}


1		<?php
2
3		namespace mpyw\Co;
4		use mpyw\Co\Internal\TypeUtils;
5		use mpyw\Co\Internal\ControlUtils;
6		use mpyw\Co\Internal\YieldableUtils;
7		use mpyw\Co\Internal\CoOption;
8		use mpyw\Co\Internal\GeneratorContainer;
9		use mpyw\Co\Internal\Pool;
10		use React\Promise\Deferred;
11		use React\Promise\PromiseInterface;
12		use React\Promise\ExtendedPromiseInterface;
13		use React\Promise\FulfilledPromise;
14		use React\Promise\RejectedPromise;
15
16		class Co implements CoInterface
17		{
18		/**
19		* Instance of myself.
20		* @var Co
21		*/
22		private static $self;
23
24		/**
25		* Options.
26		* @var CoOption
27		*/
28		private $options;
29
30		/**
31		* cURL request pool object.
32		* @var Pool
33		*/
34		private $pool;
35
36		/**
37		* Running cURL or Generator identifiers.
38		* @var array
39		*/
40		private $runners = [];
41
42		/**
43		* Overwrite CoOption default.
44		* @param array $options
45		*/
46	1	public static function setDefaultOptions(array $options)
47	1	{
48	1	CoOption::setDefault($options);
49	1	}
50
51		/**
52		* Get CoOption default as array.
53		* @return array
54		*/
55	1	public static function getDefaultOptions()
56	1	{
57	1	return CoOption::getDefault();
58		}
59
60		/**
61		* Wait until value is recursively resolved to return it.
62		* This function call must be atomic.
63		* @param mixed $value
64		* @param array $options
65		* @return mixed
66		*/
67	31	public static function wait($value, array $options = [])
68	31	{
69		try {
70	31	if (self::$self) {
71	1	throw new \BadMethodCallException('Co::wait() is already running. Use Co::async() instead.');
72		}
73	31	self::$self = new self;
74	31	self::$self->options = new CoOption($options);
75	31	self::$self->pool = new Pool(self::$self->options);
76	31	return self::$self->start($value);
77		} finally {
78	31	self::$self = null;
79		}
80		// @codeCoverageIgnoreStart
81		}
82		// @codeCoverageIgnoreEnd
83
84		/**
85		* Value is recursively resolved, but we never wait it.
86		* This function must be called along with Co::wait().
87		* @param mixed $value
88		* @param mixed $throw
89		*/
90	6	public static function async($value, $throw = null)
91	6	{
92	6	if (!self::$self) {
93	1	throw new \BadMethodCallException('Co::async() must be called along with Co::wait(). ');
94		}
95	5	if ($throw !== null) {
96	3	$throw = filter_var($throw, FILTER_VALIDATE_BOOLEAN, [
97	3	'flags' => FILTER_NULL_ON_FAILURE,
98		]);
99	3	if ($throw === null) {
100	1	throw new \InvalidArgumentException("\$throw must be null or boolean.");
101		}
102		}
103	4	self::$self->start($value, false, $throw);
104	4	}
105
106		/**
107		* Return if Co::wait() is running.
108		* @return bool
109		*/
110	2	public static function isRunning()
111	2	{
112	2	return (bool)self::$self;
113		}
114
115		/**
116		* External instantiation is forbidden.
117		*/
118		private function __construct() {}
119
120		/**
121		* Start resovling.
122		* @param mixed $value
123		* @param bool $wait
124		* @param mixed $throw Used for Co::async() overrides.
125		* @param mixed If $wait, return resolved value.
126		*/
127	31	private function start($value, $wait = true, $throw = null)
128	31	{
129	31	$return = null;
130		// For convenience, all values are wrapped into generator
131	31	$con = YieldableUtils::normalize($this->getRootGenerator($throw, $value, $return));
132	31	$promise = $this->processGeneratorContainerRunning($con);
133	27	if ($promise instanceof ExtendedPromiseInterface) {
134		// This is actually 100% true; just used for unwrapping Exception thrown.
135	27	$promise->done();
136		}
137		// We have to wait $return only if $wait
138	27	if ($wait) {
139	27	$this->pool->wait();
140	17	return $return;
141		}
142	4	}
143
144		/**
145		* Handle resolving generators.
146		* @param GeneratorContainer $gc
147		* @return PromiseInterface
148		*/
149	27	private function processGeneratorContainer(GeneratorContainer $gc)
150	27	{
151	27	return $gc->valid()
152	26	? $this->processGeneratorContainerRunning($gc)
153	27	: $this->processGeneratorContainerDone($gc);
154		}
155
156		/**
157		* Handle resolving generators already done.
158		* @param GeneratorContainer $gc
159		* @return PromiseInterface
160		*/
161	27	private function processGeneratorContainerDone(GeneratorContainer $gc)
162	27	{
163		// If exception has been thrown in generator, we have to propagate it as rejected value
164	27	if ($gc->thrown()) {
165	20	return new RejectedPromise($gc->getReturnOrThrown());
166		}
167
168		// Now we normalize returned value
169	27	$returned = YieldableUtils::normalize($gc->getReturnOrThrown(), $gc->getYieldKey());
170	27	$yieldables = YieldableUtils::getYieldables($returned, [], $this->runners);
171
172		// If normalized value contains yieldables, we have to chain resolver
173	27	if ($yieldables) {
174	4	$deferred = new Deferred;
175	4	return $this->promiseAll($yieldables, true)
176	4	->then(
177	4	YieldableUtils::getApplier($returned, $yieldables, [$deferred, 'resolve']),
178	4	[$deferred, 'reject']
179		)
180	4	->then(function () use ($yieldables, $deferred) {
181	4	$this->runners = array_diff_key($this->runners, $yieldables);
182	4	return $deferred->promise();
183	4	});
184		}
185
186		// Propagate normalized returned value
187	27	return new FulfilledPromise($returned);
188		}
189
190		/**
191		* Handle resolving generators still running.
192		* @param GeneratorContainer $gc
193		* @return PromiseInterface
194		*/
195	31	private function processGeneratorContainerRunning(GeneratorContainer $gc)
196	31	{
197		// Check delay request yields
198	31	if ($gc->key() === CoInterface::DELAY) {
199	3	return $this->pool->addDelay($gc->current())
200	3	->then(function () use ($gc) {
201	3	$gc->send(null);
202	3	return $this->processGeneratorContainer($gc);
203	3	});
204		}
205
206		// Now we normalize yielded value
207	31	$yielded = YieldableUtils::normalize($gc->current());
208	31	$yieldables = YieldableUtils::getYieldables($yielded, [], $this->runners);
209	27	if (!$yieldables) {
210		// If there are no yieldables, send yielded value back into generator
211	19	$gc->send($yielded);
212		// Continue
213	19	return $this->processGeneratorContainer($gc);
214		}
215
216		// Chain resolver
217	27	return $this->promiseAll($yieldables, $gc->key() !== CoInterface::SAFE)
218	27	->then(
219	27	YieldableUtils::getApplier($yielded, $yieldables, [$gc, 'send']),
220	27	[$gc, 'throw_']
221	27	)->then(function () use ($gc, $yieldables) {
222		// Continue
223	27	$this->runners = array_diff_key($this->runners, $yieldables);
224	27	return $this->processGeneratorContainer($gc);
225	27	});
226		}
227
228		/**
229		* Return root wrapper generator.
230		* @param mixed $throw
231		* @param mixed $value
232		* @param mixed &$return
233		*/
234	31	private function getRootGenerator($throw, $value, &$return)
235	31	{
236		try {
237	31	if ($throw !== null) {
238	2	$key = $throw ? null : CoInterface::SAFE;
239		} else {
240	31	$key = $this->options['throw'] ? null : CoInterface::SAFE;
241		}
242	31	$return = (yield $key => $value);
243	18	return;
244	11	} catch (\Throwable $e) {} catch (\Exception $e) {}
245	11	$this->pool->reserveHaltException($e);
246	11	}
247
248		/**
249		* Promise all changes in yieldables are prepared.
250		* @param array $yieldables
251		* @param bool $throw_acceptable
252		* @return PromiseInterface
253		*/
254	27	private function promiseAll(array $yieldables, $throw_acceptable)
255	27	{
256	27	$promises = [];
257	27	foreach ($yieldables as $yieldable) {
258		// Add or enqueue cURL handles
259	27	if (TypeUtils::isCurl($yieldable['value'])) {
260	10	$promises[(string)$yieldable['value']] = $this->pool->addCurl($yieldable['value']);
261	10	continue;
262		}
263		// Process generators
264	26	if (TypeUtils::isGeneratorContainer($yieldable['value'])) {
265	26	$promises[(string)$yieldable['value']] = $this->processGeneratorContainer($yieldable['value']);
266	26	continue;
267		}
268		}
269		// If caller cannot accept exception,
270		// we handle rejected value as resolved.
271	27	if (!$throw_acceptable) {
272	10	$promises = array_map(
273	10	['\mpyw\Co\Internal\YieldableUtils', 'safePromise'],
274		$promises
275		);
276		}
277	27	return \React\Promise\all($promises);
278		}
279
280		/**
281		* Wrap value with the Generator that returns the first successful result.
282		* If all yieldables failed, AllFailedException is thrown.
283		* If no yieldables found, AllFailedException is thrown.
284		*
285		* @param mixed $value
286		* @return \Generator Resolved value.
287		* @throws AllFailedException
288		*/
289	4	public static function any($value)
290	4	{
291	4	yield Co::RETURN_WITH => (yield ControlUtils::anyOrRace(
292		$value,
293	4	['\mpyw\Co\Internal\ControlUtils', 'reverse'],
294	4	'Co::any() failed.'
295		));
296		// @codeCoverageIgnoreStart
297		}
298		// @codeCoverageIgnoreEnd
299
300		/**
301		* Wrap value with the Generator that returns the first result.
302		* If no yieldables found, AllFailedException is thrown.
303		*
304		* @param mixed $value
305		* @return \Generator Resolved value.
306		* @throws \RuntimeException\|AllFailedException
307		*/
308	3	public static function race($value)
309	3	{
310	3	yield Co::RETURN_WITH => (yield ControlUtils::anyOrRace(
311		$value,
312	3	['\mpyw\Co\Internal\ControlUtils', 'fail'],
313	3	'Co::race() failed.'
314		));
315		// @codeCoverageIgnoreStart
316		}
317		// @codeCoverageIgnoreEnd
318
319		/**
320		* Wrap value with the Generator that returns the all results.
321		* Normally you don't have to use this method, just yield an array that contains yieldables.
322		* You should use only with Co::race() or Co::any().
323		*
324		* @param mixed $value
325		* @return \Generator Resolved value.
326		* @throws \RuntimeException
327		*/
328	1	public static function all($value)
329	1	{
330	1	yield Co::RETURN_WITH => (yield $value);
331		// @codeCoverageIgnoreStart
332		}
333		// @codeCoverageIgnoreEnd
334		}
335

mpyw / co

Co::processGeneratorContainerRunning() A last analyzed 2019-12-10 22:57 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

Co::processGeneratorContainerRunning() A
last analyzed 2019-12-10 22:57 UTC