NodalFlow::getStats() - Code Metrics - Inspection of "unified flow map" - fab2s/NodalFlow - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#1)

by Fabrice

created 2017-11-22 08:33 UTC

NodalFlow::getStats() A

↳ Parent: NodalFlow

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	4
c	0
b	0
f	0
rs	10
cc	1
eloc	2
nc	1
nop	0

<?php

/*
 * This file is part of NodalFlow.
 *     (c) Fabrice de Stefanis / https://github.com/fab2s/NodalFlow
 * This source file is licensed under the MIT license which you will
 * find in the LICENSE file or at https://opensource.org/licenses/MIT
 */

namespace fab2s\NodalFlow;

use fab2s\NodalFlow\Callbacks\CallbackInterface;
use fab2s\NodalFlow\Flows\FlowAbstract;
use fab2s\NodalFlow\Flows\FlowMap;
use fab2s\NodalFlow\Flows\FlowStatus;
use fab2s\NodalFlow\Flows\InterrupterInterface;
use fab2s\NodalFlow\Nodes\BranchNodeInterface;
use fab2s\NodalFlow\Nodes\ExecNodeInterface;
use fab2s\NodalFlow\Nodes\NodeInterface;
use fab2s\NodalFlow\Nodes\TraversableNodeInterface;

/**
 * Class NodalFlow
 */
class NodalFlow extends FlowAbstract
{
    /**
     * Flow steps triggering callbacks
     */
    const FLOW_START    = 'start';
    const FLOW_PROGRESS = 'progress';
    const FLOW_SUCCESS  = 'success';
    const FLOW_FAIL     = 'fail';

    /**
     * @var array
     */
    protected $flowIncrements = [];

    /**
     * The current Node index
     *
     * @var int
     */
    protected $nodeIdx = 0;

    /**
     * The last index value
     *
     * @var int
     */
    protected $lastIdx = 0;

    /**
     * The number of Node in this Flow
     *
     * @var int
     */
    protected $nodeCount = 0;

    /**
     * The number of iteration within this Flow
     *
     * @var int
     */
    protected $numIterate = 0;

    /**
     * The current registered Callback class if any
     *
     * @var CallbackInterface|null
     */
    protected $callBack;

    /**
     * Continue flag
     *
     * @var bool
     */
    protected $continue = false;

    /**
     * Break Flag
     *
     * @var bool
     */
    protected $break = false;

    /**
     * @var string|bool
     */
    protected $interruptNodeId;

    /**
     * Instantiate a Flow
     */
    public function __construct()
    {
        $this->flowMap = new FlowMap($this, $this->flowIncrements);
    }

    /**
     * Adds a Node to the flow
     *
     * @param NodeInterface $node
     *
     * @throws NodalFlowException
     *
     * @return $this
     */
    public function add(NodeInterface $node)
    {
        if ($node instanceof BranchNodeInterface) {
            // this node is a branch, set it's parent
            $node->getPayload()->setParent($this);
        }

        $node->setCarrier($this);

        $this->flowMap->register($node, $this->nodeIdx);
        $this->nodes[$this->nodeIdx] = $node;

        ++$this->nodeIdx;

        return $this;
    }

    /**
     * Adds a Payload Node to the Flow
     *
     * @param callable $payload
     * @param mixed    $isAReturningVal
     * @param mixed    $isATraversable
     *
     * @return $this
     */
    public function addPayload(callable $payload, $isAReturningVal, $isATraversable = false)
    {
        $node = PayloadNodeFactory::create($payload, $isAReturningVal, $isATraversable);

        $this->add($node);

        return $this;
    }

    /**
     * Register callback class
     *
     * @param CallbackInterface $callBack
     *
     * @return $this
     */
    public function setCallBack(CallbackInterface $callBack)
    {
        $this->callBack = $callBack;

        return $this;
    }

    /**
     * Used to set the eventual Node Target of an Interrupt signal
     * set to :
     * - A node hash to target
     * - true to interrupt every upstream nodes
     *     in this Flow
     * - false to only interrupt up to the first
     *     upstream Traversable in this Flow
     *
     * @param string|bool $interruptNodeId
     *
     * @return $this
     */
    public function setInterruptNodeId($interruptNodeId)
    {
        $this->interruptNodeId = $interruptNodeId;

        return $this;
    }

    /**
     * Execute the flow
     *
     * @param null|mixed $param The eventual init argument to the first node
     *                          or, in case of a branch, the last relevant
     *                          argument from upstream Flow
     *
     * @throws NodalFlowException
     *
     * @return mixed the last result of the
     *               last returning value node
     */
    public function exec($param = null)
    {
        try {
            $result = $this->rewind()
                    ->flowStart()
                    ->recurse($param);

            // set flowStatus to make sure that we have the proper
            // value in flowEnd even when overridden without (or when
            // improperly) calling parent
            if ($this->flowStatus->isRunning()) {
                $this->flowStatus = new FlowStatus(FlowStatus::FLOW_CLEAN);
            }

            $this->flowEnd();

            return $result;
        } catch (\Exception $e) {
            $this->flowStatus = new FlowStatus(FlowStatus::FLOW_EXCEPTION);
            $this->flowEnd();
            if ($e instanceof NodalFlowException) {
                throw $e;
            }

            throw new NodalFlowException('Flow execution failed', 0, $e, [
                'nodeMap' => $this->getNodeMap(),
            ]);
        }
    }

    /**
     * Rewinds the Flow
     *
     * @return $this
     */
    public function rewind()
    {
        $this->nodeCount       = count($this->nodes);
        $this->lastIdx         = $this->nodeCount - 1;
        $this->break           = false;
        $this->continue        = false;
        $this->interruptNodeId = null;

        return $this;
    }

    /**
     * Define the progress modulo, Progress Callback will be
     * triggered upon each iteration in the flow modulo $progressMod
     *
     * @param int $progressMod
     *
     * @return $this
     */
    public function setProgressMod($progressMod)
    {
        $this->progressMod = max(1, (int) $progressMod);

        return $this;
    }

    /**
     * Break the flow's execution, conceptually similar to breaking
     * a regular loop
     *
     * @param InterrupterInterface|null $flowInterrupt
     *
     * @return $this
     */
    public function breakFlow(InterrupterInterface $flowInterrupt = null)
    {
        return $this->interruptFlow(InterrupterInterface::TYPE_BREAK, $flowInterrupt);
    }

    /**
     * Continue the flow's execution, conceptually similar to continuing
     * a regular loop
     *
     * @param InterrupterInterface|null $flowInterrupt
     *
     * @return $this
     */
    public function continueFlow(InterrupterInterface $flowInterrupt = null)
    {
        return $this->interruptFlow(InterrupterInterface::TYPE_CONTINUE, $flowInterrupt);
    }

    /**
     * @param string                    $interruptType
     * @param InterrupterInterface|null $flowInterrupt
     *
     * @throws NodalFlowException
     *
     * @return $this
     */
    public function interruptFlow($interruptType, InterrupterInterface $flowInterrupt = null)
    {
        switch ($interruptType) {
            case InterrupterInterface::TYPE_CONTINUE:
                $this->continue = true;
                $this->flowMap->incrementFlow('num_continue');
                break;
            case InterrupterInterface::TYPE_BREAK:
                $this->flowStatus = new FlowStatus(FlowStatus::FLOW_DIRTY);
                $this->break      = true;
                $this->flowMap->incrementFlow('num_break');
                break;
            default:
                throw new NodalFlowException('FlowInterrupt Type missing');
        }

        if ($flowInterrupt) {
            $flowInterrupt->setType($interruptType)->propagate($this);
        }

        return $this;
    }

    /**
     * @param NodeInterface $node
     *
     * @return bool
     */
    protected function interruptNode(NodeInterface $node)
    {
        // if we have an interruptNodeId, bubble up until we match a node
        // else stop propagation
        return $this->interruptNodeId ? $this->interruptNodeId !== $node->getId() : false;
    }

    /**
     * Triggered just before the flow starts
     *
     * @return $this
     */
    protected function flowStart()
    {
        $this->flowMap->incrementFlow('num_exec')->flowStart();
        $this->triggerCallback(static::FLOW_START);

        // flow is started
        $this->flowStatus = new FlowStatus(FlowStatus::FLOW_RUNNING);

        return $this;
    }

    /**
     * Triggered right after the flow stops
     *
     * @return $this
     */
    protected function flowEnd()
    {
        $this->flowMap->flowEnd();

        $this->triggerCallback($this->flowStatus->isException() ? static::FLOW_FAIL : static::FLOW_SUCCESS);

        return $this;
    }

    /**
     * Recurse over nodes which may as well be Flows and
     * Traversable ...
     * Welcome to the abysses of recursion or iter-recursion ^^
     *
     * `recurse` perform kind of an hybrid recursion as the
     * Flow is effectively iterating and recurring over its
     * Nodes, which may as well be seen as over itself
     *
     * Iterating tends to limit the amount of recursion levels:
     * recursion is only triggered when executing a Traversable
     * Node's downstream Nodes while every consecutive exec
     * Nodes are executed within a while loop.
     * And recursion keeps the size of the recursion context
     * to a minimum as pretty much everything is done by the
     * iterating instance
     *
     * @param mixed $param
     * @param int   $nodeIdx
     *
     * @return mixed the last value returned by the last
     *               returning value Node in the flow
     */
    protected function recurse($param = null, $nodeIdx = 0)
    {
        while ($nodeIdx <= $this->lastIdx) {
            $node      = $this->nodes[$nodeIdx];
            $nodeStats = &$this->flowMap->getNodeStat($node->getId());

            $returnVal = $node->isReturningVal();

            if ($node->isTraversable()) {
                /** @var TraversableNodeInterface $node */
                foreach ($node->getTraversable($param) as $value) {
                    if ($returnVal) {
                        // pass current $value as next param
                        $param = $value;
                    }

                    ++$nodeStats['num_iterate'];
                    ++$this->numIterate;
                    if (!($this->numIterate % $this->progressMod)) {
                        $this->triggerCallback(static::FLOW_PROGRESS, $node);
                    }

                    $param = $this->recurse($param, $nodeIdx + 1);
                    if ($this->continue) {
                        if ($this->continue = $this->interruptNode($node)) {
                            // since we want to bubble the continue upstream
                            // we break here waiting for next $param if any
                            ++$nodeStats['num_break'];
                            break;
                        }

                        // we drop one iteration
                        ++$nodeStats['num_continue'];
                        continue;
                    }

                    if ($this->break) {
                        // we drop all subsequent iterations
                        ++$nodeStats['num_break'];
                        $this->break = $this->interruptNode($node);
                        break;
                    }
                }

                // we reached the end of this Traversable and executed all its downstream Nodes
                ++$nodeStats['num_exec'];

                return $param;
            }

            /** @var ExecNodeInterface $node */
            $value = $node->exec($param);
            ++$nodeStats['num_exec'];

            if ($this->continue) {
                ++$nodeStats['num_continue'];
                // a continue does not need to bubble up unless
                // it specifically targets a node in this flow
                // or targets an upstream flow
                $this->continue = $this->interruptNode($node);

                return $param;
            }

            if ($this->break) {
                ++$nodeStats['num_break'];
                // a break always need to bubble up to the first upstream Traversable if any
                return $param;
            }

            if ($returnVal) {
                // pass current $value as next param
                $param = $value;
            }

            ++$nodeIdx;
        }

        // we reached the end of this recursion
        return $param;
    }

    /**
     * KISS helper to trigger Callback slots
     *
     * @param string             $which
     * @param null|NodeInterface $node
     *
     * @return $this
     */
    protected function triggerCallback($which, NodeInterface $node = null)
    {
        if (null !== $this->callBack) {
            $this->callBack->$which($this, $node);
        }

        return $this;
    }
}


1			<?php
2
3			/*
4			* This file is part of NodalFlow.
5			* (c) Fabrice de Stefanis / https://github.com/fab2s/NodalFlow
6			* This source file is licensed under the MIT license which you will
7			* find in the LICENSE file or at https://opensource.org/licenses/MIT
8			*/
9
10			namespace fab2s\NodalFlow;
11
12			use fab2s\NodalFlow\Callbacks\CallbackInterface;
13			use fab2s\NodalFlow\Flows\FlowAbstract;
14			use fab2s\NodalFlow\Flows\FlowMap;
15			use fab2s\NodalFlow\Flows\FlowStatus;
16			use fab2s\NodalFlow\Flows\InterrupterInterface;
17			use fab2s\NodalFlow\Nodes\BranchNodeInterface;
18			use fab2s\NodalFlow\Nodes\ExecNodeInterface;
19			use fab2s\NodalFlow\Nodes\NodeInterface;
20			use fab2s\NodalFlow\Nodes\TraversableNodeInterface;
21
22			/**
23			* Class NodalFlow
24			*/
25			class NodalFlow extends FlowAbstract
26			{
27			/**
28			* Flow steps triggering callbacks
29			*/
30			const FLOW_START = 'start';
31			const FLOW_PROGRESS = 'progress';
32			const FLOW_SUCCESS = 'success';
33			const FLOW_FAIL = 'fail';
34
35			/**
36			* @var array
37			*/
38			protected $flowIncrements = [];
39
40			/**
41			* The current Node index
42			*
43			* @var int
44			*/
45			protected $nodeIdx = 0;
46
47			/**
48			* The last index value
49			*
50			* @var int
51			*/
52			protected $lastIdx = 0;
53
54			/**
55			* The number of Node in this Flow
56			*
57			* @var int
58			*/
59			protected $nodeCount = 0;
60
61			/**
62			* The number of iteration within this Flow
63			*
64			* @var int
65			*/
66			protected $numIterate = 0;
67
68			/**
69			* The current registered Callback class if any
70			*
71			* @var CallbackInterface\|null
72			*/
73			protected $callBack;
74
75			/**
76			* Continue flag
77			*
78			* @var bool
79			*/
80			protected $continue = false;
81
82			/**
83			* Break Flag
84			*
85			* @var bool
86			*/
87			protected $break = false;
88
89			/**
90			* @var string\|bool
91			*/
92			protected $interruptNodeId;
93
94			/**
95			* Instantiate a Flow
96			*/
97			public function __construct()
98			{
99			$this->flowMap = new FlowMap($this, $this->flowIncrements);
100			}
101
102			/**
103			* Adds a Node to the flow
104			*
105			* @param NodeInterface $node
106			*
107			* @throws NodalFlowException
108			*
109			* @return $this
110			*/
111			public function add(NodeInterface $node)
112			{
113			if ($node instanceof BranchNodeInterface) {
114			// this node is a branch, set it's parent
115			$node->getPayload()->setParent($this);
116			}
117
118			$node->setCarrier($this);
119
120			$this->flowMap->register($node, $this->nodeIdx);
121			$this->nodes[$this->nodeIdx] = $node;
122
123			++$this->nodeIdx;
124
125			return $this;
126			}
127
128			/**
129			* Adds a Payload Node to the Flow
130			*
131			* @param callable $payload
132			* @param mixed $isAReturningVal
133			* @param mixed $isATraversable
134			*
135			* @return $this
136			*/
137			public function addPayload(callable $payload, $isAReturningVal, $isATraversable = false)
138			{
139			$node = PayloadNodeFactory::create($payload, $isAReturningVal, $isATraversable);
140
141			$this->add($node);
142
143			return $this;
144			}
145
146			/**
147			* Register callback class
148			*
149			* @param CallbackInterface $callBack
150			*
151			* @return $this
152			*/
153			public function setCallBack(CallbackInterface $callBack)
154			{
155			$this->callBack = $callBack;
156
157			return $this;
158			}
159
160			/**
161			* Used to set the eventual Node Target of an Interrupt signal
162			* set to :
163			* - A node hash to target
164			* - true to interrupt every upstream nodes
165			* in this Flow
166			* - false to only interrupt up to the first
167			* upstream Traversable in this Flow
168			*
169			* @param string\|bool $interruptNodeId
170			*
171			* @return $this
172			*/
173			public function setInterruptNodeId($interruptNodeId)
174			{
175			$this->interruptNodeId = $interruptNodeId;
176
177			return $this;
178			}
179
180			/**
181			* Execute the flow
182			*
183			* @param null\|mixed $param The eventual init argument to the first node
184			* or, in case of a branch, the last relevant
185			* argument from upstream Flow
186			*
187			* @throws NodalFlowException
188			*
189			* @return mixed the last result of the
190			* last returning value node
191			*/
192			public function exec($param = null)
193			{
194			try {
195			$result = $this->rewind()
196			->flowStart()
197			->recurse($param);
198
199			// set flowStatus to make sure that we have the proper
200			// value in flowEnd even when overridden without (or when
201			// improperly) calling parent
202			if ($this->flowStatus->isRunning()) {
203			$this->flowStatus = new FlowStatus(FlowStatus::FLOW_CLEAN);
204			}
205
206			$this->flowEnd();
207
208			return $result;
209			} catch (\Exception $e) {
210			$this->flowStatus = new FlowStatus(FlowStatus::FLOW_EXCEPTION);
211			$this->flowEnd();
212			if ($e instanceof NodalFlowException) {
213			throw $e;
214			}
215
216			throw new NodalFlowException('Flow execution failed', 0, $e, [
217			'nodeMap' => $this->getNodeMap(),
218			]);
219			}
220			}
221
222			/**
223			* Rewinds the Flow
224			*
225			* @return $this
226			*/
227			public function rewind()
228			{
229			$this->nodeCount = count($this->nodes);
230			$this->lastIdx = $this->nodeCount - 1;
231			$this->break = false;
232			$this->continue = false;
233			$this->interruptNodeId = null;
234
235			return $this;
236			}
237
238			/**
239			* Define the progress modulo, Progress Callback will be
240			* triggered upon each iteration in the flow modulo $progressMod
241			*
242			* @param int $progressMod
243			*
244			* @return $this
245			*/
246			public function setProgressMod($progressMod)
247			{
248			$this->progressMod = max(1, (int) $progressMod);
249
250			return $this;
251			}
252
253			/**
254			* Break the flow's execution, conceptually similar to breaking
255			* a regular loop
256			*
257			* @param InterrupterInterface\|null $flowInterrupt
258			*
259			* @return $this
260			*/
261			public function breakFlow(InterrupterInterface $flowInterrupt = null)
262			{
263			return $this->interruptFlow(InterrupterInterface::TYPE_BREAK, $flowInterrupt);
264			}
265
266			/**
267			* Continue the flow's execution, conceptually similar to continuing
268			* a regular loop
269			*
270			* @param InterrupterInterface\|null $flowInterrupt
271			*
272			* @return $this
273			*/
274			public function continueFlow(InterrupterInterface $flowInterrupt = null)
275			{
276			return $this->interruptFlow(InterrupterInterface::TYPE_CONTINUE, $flowInterrupt);
277			}
278
279			/**
280			* @param string $interruptType
281			* @param InterrupterInterface\|null $flowInterrupt
282			*
283			* @throws NodalFlowException
284			*
285			* @return $this
286			*/
287			public function interruptFlow($interruptType, InterrupterInterface $flowInterrupt = null)
288			{
289			switch ($interruptType) {
290			case InterrupterInterface::TYPE_CONTINUE:
291			$this->continue = true;
292			$this->flowMap->incrementFlow('num_continue');
293			break;
294			case InterrupterInterface::TYPE_BREAK:
295			$this->flowStatus = new FlowStatus(FlowStatus::FLOW_DIRTY);
296			$this->break = true;
297			$this->flowMap->incrementFlow('num_break');
298			break;
299			default:
300			throw new NodalFlowException('FlowInterrupt Type missing');
301			}
302
303			if ($flowInterrupt) {
304			$flowInterrupt->setType($interruptType)->propagate($this);
305			}
306
307			return $this;
308			}
309
310			/**
311			* @param NodeInterface $node
312			*
313			* @return bool
314			*/
315			protected function interruptNode(NodeInterface $node)
316			{
317			// if we have an interruptNodeId, bubble up until we match a node
318			// else stop propagation
319			return $this->interruptNodeId ? $this->interruptNodeId !== $node->getId() : false;
320			}
321
322			/**
323			* Triggered just before the flow starts
324			*
325			* @return $this
326			*/
327			protected function flowStart()
328			{
329			$this->flowMap->incrementFlow('num_exec')->flowStart();
330			$this->triggerCallback(static::FLOW_START);
331
332			// flow is started
333			$this->flowStatus = new FlowStatus(FlowStatus::FLOW_RUNNING);
334
335			return $this;
336			}
337
338			/**
339			* Triggered right after the flow stops
340			*
341			* @return $this
342			*/
343			protected function flowEnd()
344			{
345			$this->flowMap->flowEnd();
346
347			$this->triggerCallback($this->flowStatus->isException() ? static::FLOW_FAIL : static::FLOW_SUCCESS);
348
349			return $this;
350			}
351
352			/**
353			* Recurse over nodes which may as well be Flows and
354			* Traversable ...
355			* Welcome to the abysses of recursion or iter-recursion ^^
356			*
357			* `recurse` perform kind of an hybrid recursion as the
358			* Flow is effectively iterating and recurring over its
359			* Nodes, which may as well be seen as over itself
360			*
361			* Iterating tends to limit the amount of recursion levels:
362			* recursion is only triggered when executing a Traversable
363			* Node's downstream Nodes while every consecutive exec
364			* Nodes are executed within a while loop.
365			* And recursion keeps the size of the recursion context
366			* to a minimum as pretty much everything is done by the
367			* iterating instance
368			*
369			* @param mixed $param
370			* @param int $nodeIdx
371			*
372			* @return mixed the last value returned by the last
373			* returning value Node in the flow
374			*/
375			protected function recurse($param = null, $nodeIdx = 0)
376			{
377			while ($nodeIdx <= $this->lastIdx) {
378			$node = $this->nodes[$nodeIdx];
379			$nodeStats = &$this->flowMap->getNodeStat($node->getId());
380
381			$returnVal = $node->isReturningVal();
382
383			if ($node->isTraversable()) {
384			/** @var TraversableNodeInterface $node */
385			foreach ($node->getTraversable($param) as $value) {
386			if ($returnVal) {
387			// pass current $value as next param
388			$param = $value;
389			}
390
391			++$nodeStats['num_iterate'];
392			++$this->numIterate;
393			if (!($this->numIterate % $this->progressMod)) {
394			$this->triggerCallback(static::FLOW_PROGRESS, $node);
395			}
396
397			$param = $this->recurse($param, $nodeIdx + 1);
398			if ($this->continue) {
399			if ($this->continue = $this->interruptNode($node)) {
400			// since we want to bubble the continue upstream
401			// we break here waiting for next $param if any
402			++$nodeStats['num_break'];
403			break;
404			}
405
406			// we drop one iteration
407			++$nodeStats['num_continue'];
408			continue;
409			}
410
411			if ($this->break) {
412			// we drop all subsequent iterations
413			++$nodeStats['num_break'];
414			$this->break = $this->interruptNode($node);
415			break;
416			}
417			}
418
419			// we reached the end of this Traversable and executed all its downstream Nodes
420			++$nodeStats['num_exec'];
421
422			return $param;
423			}
424
425			/** @var ExecNodeInterface $node */
426			$value = $node->exec($param);
427			++$nodeStats['num_exec'];
428
429			if ($this->continue) {
430			++$nodeStats['num_continue'];
431			// a continue does not need to bubble up unless
432			// it specifically targets a node in this flow
433			// or targets an upstream flow
434			$this->continue = $this->interruptNode($node);
435
436			return $param;
437			}
438
439			if ($this->break) {
440			++$nodeStats['num_break'];
441			// a break always need to bubble up to the first upstream Traversable if any
442			return $param;
443			}
444
445			if ($returnVal) {
446			// pass current $value as next param
447			$param = $value;
448			}
449
450			++$nodeIdx;
451			}
452
453			// we reached the end of this recursion
454			return $param;
455			}
456
457			/**
458			* KISS helper to trigger Callback slots
459			*
460			* @param string $which
461			* @param null\|NodeInterface $node
462			*
463			* @return $this
464			*/
465			protected function triggerCallback($which, NodeInterface $node = null)
466			{
467			if (null !== $this->callBack) {
468			$this->callBack->$which($this, $node);
469			}
470
471			return $this;
472			}
473			}
474

fab2s / NodalFlow

Pull Request — master (#1)

NodalFlow::getStats() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like