NodalFlow::onFlowStart() - Code Metrics - Inspection of "On flow start" - fab2s/NodalFlow - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — onFlowStart ( 676580...51dac1 )

by Fabrice

created 2023-02-03 19:46 UTC

NodalFlow::onFlowStart() A

↳ Parent: NodalFlow

Complexity

Conditions	3
Paths	3

Size

Total Lines	9
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	4
nc	3
nop	0
dl	0
loc	9
rs	10
c	0
b	0
f	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\Events\FlowEventInterface;
use fab2s\NodalFlow\Flows\FlowAbstract;
use fab2s\NodalFlow\Flows\FlowInterface;
use fab2s\NodalFlow\Flows\FlowMap;
use fab2s\NodalFlow\Flows\FlowRegistry;
use fab2s\NodalFlow\Flows\FlowStatus;
use fab2s\NodalFlow\Nodes\Interfaces\BranchNodeInterface;
filter:
    dependency_paths: ["lib/*"]
use fab2s\NodalFlow\Nodes\Interfaces\ExecNodeInterface;
filter:
    dependency_paths: ["lib/*"]
use fab2s\NodalFlow\Nodes\Interfaces\NodeInterface;
filter:
    dependency_paths: ["lib/*"]
use fab2s\NodalFlow\Nodes\Interfaces\OnFlowStartInterface;
use fab2s\NodalFlow\Nodes\Interfaces\TraversableNodeInterface;
filter:
    dependency_paths: ["lib/*"]
use Throwable;

/**
 * Class NodalFlow
 */
class NodalFlow extends FlowAbstract
{
    /**
     * @var array
     */
    protected $flowIncrements = [];

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

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

    /**
     * Adds a Node to the flow
     *
     * @param NodeInterface $node
     *
     * @throws NodalFlowException
     *
     * @return $this
     */
    public function add(NodeInterface $node): FlowInterface
    {
        if ($node instanceof BranchNodeInterface) {
            // this node is a branch
            $childFlow = $node->getPayload();
            $this->branchFlowCheck($childFlow);
            $childFlow->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
     *
     * @throws NodalFlowException
     *
     * @return $this
     */
    public function addPayload(callable $payload, bool $isAReturningVal, bool $isATraversable = false): FlowInterface
    {
        $node = PayloadNodeFactory::create($payload, $isAReturningVal, $isATraversable);

        $this->add($node);

        return $this;
    }

    /**
     * Replaces a node with another one
     *
     * @param int           $nodeIdx
     * @param NodeInterface $node
     *
     * @throws NodalFlowException
     *
     * @return static
     */
    public function replace(int $nodeIdx, NodeInterface $node): FlowInterface
    {
        if (!isset($this->nodes[$nodeIdx])) {
            throw new NodalFlowException('Argument 1 should be a valid index in nodes', 1, null, [
                'nodeIdx' => $nodeIdx,
                'node'    => get_class($node),
            ]);
        }

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

        return $this;
    }

    /**
     * @param string|null $nodeId
     * @param mixed|null  $param
     *
     * @throws Throwable
     * @throws NodalFlowException
     *
     * @return mixed
     */
    public function sendTo(string $nodeId = null, $param = null)
    {
        $nodeIndex = 0;
        if ($nodeId !== null) {
            if (!($nodeIndex = $this->flowMap->getNodeIndex($nodeId))) {
                throw new NodalFlowException('Cannot sendTo without valid Node target', 1, null, [
                    'flowId' => $this->getId(),
                    'nodeId' => $nodeId,
                ]);
            }
        }

        return $this->rewind()->recurse($param, $nodeIndex);
    }

    /**
     * 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|Throwable
     *
     * @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 (Throwable $e) {
            $this->flowStatus = new FlowStatus(FlowStatus::FLOW_EXCEPTION, $e);
            $this->flowEnd();

            throw $e;
        }
    }

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

        return $this;
    }

    /**
     * @param FlowInterface $flow
     *
     * @throws NodalFlowException
     */
    protected function branchFlowCheck(FlowInterface $flow)
    {
        if (
            // this flow has parent already
            $flow->hasParent() ||
            // adding root flow in itself
            $this->getRootFlow($flow)->getId() === $this->getRootFlow($this)->getId()
        ) {
            throw new NodalFlowException('Cannot reuse Flow within Branches', 1, null, [
                'flowId'             => $this->getId(),
                'BranchFlowId'       => $flow->getId(),
                'BranchFlowParentId' => $flow->hasParent() ? $flow->getParent()->getId() : null,
            ]);
        }
    }

    /**
     * Triggered just before the flow starts
     *
     *
     * @return $this
     */
    protected function flowStart(): self
    {
        $this->flowMap->incrementFlow('num_exec')->flowStart();
        $this->onFlowStart()
            ->listActiveEvent(!$this->hasParent())
            ->triggerEvent(FlowEventInterface::FLOW_START);
        // flow started status kicks in after Event start to hint eventual
        // children this way, root flow is only running when a record hits
        // a branch and triggers a child flow flowStart() call
        $this->flowStatus = new FlowStatus(FlowStatus::FLOW_RUNNING);

        return $this;
    }

    protected function onFlowStart(): self
    {
        foreach ($this->nodes as $node) {
            if ($node instanceof OnFlowStartInterface) {
                $this->onFlowStart();
            }
        }

        return $this;
    }

    /**
     * Triggered right after the flow stops
     *
     * @return $this
     */
    protected function flowEnd(): self
    {
        $this->flowMap->flowEnd();
        $eventName = FlowEventInterface::FLOW_SUCCESS;
        $node      = null;
        if ($this->flowStatus->isException()) {
            $eventName = FlowEventInterface::FLOW_FAIL;
            $node      = $this->nodes[$this->nodeIdx];
        }

        // restore nodeIdx
        $this->nodeIdx = $this->lastIdx + 1;
        $this->triggerEvent($eventName, $node);

        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 the while loop.
     * The size of the recursion context is kept 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, int $nodeIdx = 0)
    {
        while ($nodeIdx <= $this->lastIdx) {
            $node          = $this->nodes[$nodeIdx];
            $this->nodeIdx = $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'];
                    if (!($nodeStats['num_iterate'] % $this->progressMod)) {
                        $this->triggerEvent(FlowEventInterface::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;
    }
}


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\Events\FlowEventInterface;
13			use fab2s\NodalFlow\Flows\FlowAbstract;
14			use fab2s\NodalFlow\Flows\FlowInterface;
15			use fab2s\NodalFlow\Flows\FlowMap;
16			use fab2s\NodalFlow\Flows\FlowRegistry;
17			use fab2s\NodalFlow\Flows\FlowStatus;
18			use fab2s\NodalFlow\Nodes\Interfaces\BranchNodeInterface;
			0 ignored issues – show Bug introduced 2023-02-03 19:48 UTC by Report Bug Copy Issue Report The type `fab2s\NodalFlow\Nodes\In...ces\BranchNodeInterface` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
19			use fab2s\NodalFlow\Nodes\Interfaces\ExecNodeInterface;
			0 ignored issues – show Bug introduced 2023-02-03 19:48 UTC by Report Bug Copy Issue Report The type `fab2s\NodalFlow\Nodes\Interfaces\ExecNodeInterface` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
20			use fab2s\NodalFlow\Nodes\Interfaces\NodeInterface;
			0 ignored issues – show Bug introduced 2023-02-03 19:48 UTC by Report Bug Copy Issue Report The type `fab2s\NodalFlow\Nodes\Interfaces\NodeInterface` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
21			use fab2s\NodalFlow\Nodes\Interfaces\OnFlowStartInterface;
22			use fab2s\NodalFlow\Nodes\Interfaces\TraversableNodeInterface;
			0 ignored issues – show Bug introduced 2023-02-03 19:48 UTC by Report Bug Copy Issue Report The type `fab2s\NodalFlow\Nodes\In...raversableNodeInterface` was not found. Maybe you did not declare it correctly or list all dependencies? The issue could also be caused by a filter entry in the build configuration. If the path has been excluded in your configuration, e.g. `excluded_paths: ["lib/"]`, you can move it to the dependency path list as follows: filter: dependency_paths: ["lib/"] For further information see https://scrutinizer-ci.com/docs/tools/php/php-scrutinizer/#list-dependency-paths Loading history...
23			use Throwable;
24
25			/**
26			* Class NodalFlow
27			*/
28			class NodalFlow extends FlowAbstract
29			{
30			/**
31			* @var array
32			*/
33			protected $flowIncrements = [];
34
35			/**
36			* The number of Node in this Flow
37			*
38			* @var int
39			*/
40			protected $nodeCount = 0;
41
42			/**
43			* Instantiate a Flow
44			*
45			* @throws NodalFlowException
46			*/
47			public function __construct()
48			{
49			$this->flowMap = new FlowMap($this, $this->flowIncrements);
50			$this->registry = new FlowRegistry;
51			}
52
53			/**
54			* Adds a Node to the flow
55			*
56			* @param NodeInterface $node
57			*
58			* @throws NodalFlowException
59			*
60			* @return $this
61			*/
62			public function add(NodeInterface $node): FlowInterface
63			{
64			if ($node instanceof BranchNodeInterface) {
65			// this node is a branch
66			$childFlow = $node->getPayload();
67			$this->branchFlowCheck($childFlow);
68			$childFlow->setParent($this);
69			}
70
71			$node->setCarrier($this);
72
73			$this->flowMap->register($node, $this->nodeIdx);
74			$this->nodes[$this->nodeIdx] = $node;
75
76			++$this->nodeIdx;
77
78			return $this;
79			}
80
81			/**
82			* Adds a Payload Node to the Flow
83			*
84			* @param callable $payload
85			* @param mixed $isAReturningVal
86			* @param mixed $isATraversable
87			*
88			* @throws NodalFlowException
89			*
90			* @return $this
91			*/
92			public function addPayload(callable $payload, bool $isAReturningVal, bool $isATraversable = false): FlowInterface
93			{
94			$node = PayloadNodeFactory::create($payload, $isAReturningVal, $isATraversable);
95
96			$this->add($node);
97
98			return $this;
99			}
100
101			/**
102			* Replaces a node with another one
103			*
104			* @param int $nodeIdx
105			* @param NodeInterface $node
106			*
107			* @throws NodalFlowException
108			*
109			* @return static
110			*/
111			public function replace(int $nodeIdx, NodeInterface $node): FlowInterface
112			{
113			if (!isset($this->nodes[$nodeIdx])) {
114			throw new NodalFlowException('Argument 1 should be a valid index in nodes', 1, null, [
115			'nodeIdx' => $nodeIdx,
116			'node' => get_class($node),
117			]);
118			}
119
120			$node->setCarrier($this);
121			$this->nodes[$nodeIdx] = $node;
122			$this->flowMap->register($node, $nodeIdx, true);
123
124			return $this;
125			}
126
127			/**
128			* @param string\|null $nodeId
129			* @param mixed\|null $param
130			*
131			* @throws Throwable
132			* @throws NodalFlowException
133			*
134			* @return mixed
135			*/
136			public function sendTo(string $nodeId = null, $param = null)
137			{
138			$nodeIndex = 0;
139			if ($nodeId !== null) {
140			if (!($nodeIndex = $this->flowMap->getNodeIndex($nodeId))) {
141			throw new NodalFlowException('Cannot sendTo without valid Node target', 1, null, [
142			'flowId' => $this->getId(),
143			'nodeId' => $nodeId,
144			]);
145			}
146			}
147
148			return $this->rewind()->recurse($param, $nodeIndex);
149			}
150
151			/**
152			* Execute the flow
153			*
154			* @param null\|mixed $param The eventual init argument to the first node
155			* or, in case of a branch, the last relevant
156			* argument from upstream Flow
157			*
158			*@throws NodalFlowException\|Throwable
159			*
160			* @return mixed the last result of the
161			* last returning value node
162			*/
163			public function exec($param = null)
164			{
165			try {
166			$result = $this->rewind()
167			->flowStart()
168			->recurse($param);
169
170			// set flowStatus to make sure that we have the proper
171			// value in flowEnd even when overridden without (or when
172			// improperly) calling parent
173			if ($this->flowStatus->isRunning()) {
174			$this->flowStatus = new FlowStatus(FlowStatus::FLOW_CLEAN);
175			}
176
177			$this->flowEnd();
178
179			return $result;
180			} catch (Throwable $e) {
181			$this->flowStatus = new FlowStatus(FlowStatus::FLOW_EXCEPTION, $e);
182			$this->flowEnd();
183
184			throw $e;
185			}
186			}
187
188			/**
189			* Rewinds the Flow
190			*
191			* @return $this
192			*/
193			public function rewind(): FlowInterface
194			{
195			$this->nodeCount = count($this->nodes);
196			$this->lastIdx = $this->nodeCount - 1;
197			$this->break = false;
198			$this->continue = false;
199			$this->interruptNodeId = null;
200
201			return $this;
202			}
203
204			/**
205			* @param FlowInterface $flow
206			*
207			* @throws NodalFlowException
208			*/
209			protected function branchFlowCheck(FlowInterface $flow)
210			{
211			if (
212			// this flow has parent already
213			$flow->hasParent() \|\|
214			// adding root flow in itself
215			$this->getRootFlow($flow)->getId() === $this->getRootFlow($this)->getId()
216			) {
217			throw new NodalFlowException('Cannot reuse Flow within Branches', 1, null, [
218			'flowId' => $this->getId(),
219			'BranchFlowId' => $flow->getId(),
220			'BranchFlowParentId' => $flow->hasParent() ? $flow->getParent()->getId() : null,
221			]);
222			}
223			}
224
225			/**
226			* Triggered just before the flow starts
227			*
228			*
229			* @return $this
230			*/
231			protected function flowStart(): self
232			{
233			$this->flowMap->incrementFlow('num_exec')->flowStart();
234			$this->onFlowStart()
235			->listActiveEvent(!$this->hasParent())
236			->triggerEvent(FlowEventInterface::FLOW_START);
237			// flow started status kicks in after Event start to hint eventual
238			// children this way, root flow is only running when a record hits
239			// a branch and triggers a child flow flowStart() call
240			$this->flowStatus = new FlowStatus(FlowStatus::FLOW_RUNNING);
241
242			return $this;
243			}
244
245			protected function onFlowStart(): self
246			{
247			foreach ($this->nodes as $node) {
248			if ($node instanceof OnFlowStartInterface) {
249			$this->onFlowStart();
250			}
251			}
252
253			return $this;
254			}
255
256			/**
257			* Triggered right after the flow stops
258			*
259			* @return $this
260			*/
261			protected function flowEnd(): self
262			{
263			$this->flowMap->flowEnd();
264			$eventName = FlowEventInterface::FLOW_SUCCESS;
265			$node = null;
266			if ($this->flowStatus->isException()) {
267			$eventName = FlowEventInterface::FLOW_FAIL;
268			$node = $this->nodes[$this->nodeIdx];
269			}
270
271			// restore nodeIdx
272			$this->nodeIdx = $this->lastIdx + 1;
273			$this->triggerEvent($eventName, $node);
274
275			return $this;
276			}
277
278			/**
279			* Recurse over nodes which may as well be Flows and Traversable ...
280			* Welcome to the abysses of recursion or iter-recursion ^^
281			*
282			* `recurse` perform kind of an hybrid recursion as the Flow
283			* is effectively iterating and recurring over its Nodes,
284			* which may as well be seen as over itself
285			*
286			* Iterating tends to limit the amount of recursion levels:
287			* recursion is only triggered when executing a Traversable
288			* Node's downstream Nodes while every consecutive exec
289			* Nodes are executed within the while loop.
290			* The size of the recursion context is kept to a minimum
291			* as pretty much everything is done by the iterating instance
292			*
293			* @param mixed $param
294			* @param int $nodeIdx
295			*
296			* @return mixed the last value returned by the last
297			* returning value Node in the flow
298			*/
299			protected function recurse($param = null, int $nodeIdx = 0)
300			{
301			while ($nodeIdx <= $this->lastIdx) {
302			$node = $this->nodes[$nodeIdx];
303			$this->nodeIdx = $nodeIdx;
304			$nodeStats = &$this->flowMap->getNodeStat($node->getId());
305			$returnVal = $node->isReturningVal();
306
307			if ($node->isTraversable()) {
308			/** @var TraversableNodeInterface $node */
309			foreach ($node->getTraversable($param) as $value) {
310			if ($returnVal) {
311			// pass current $value as next param
312			$param = $value;
313			}
314
315			++$nodeStats['num_iterate'];
316			if (!($nodeStats['num_iterate'] % $this->progressMod)) {
317			$this->triggerEvent(FlowEventInterface::FLOW_PROGRESS, $node);
318			}
319
320			$param = $this->recurse($param, $nodeIdx + 1);
321			if ($this->continue) {
322			if ($this->continue = $this->interruptNode($node)) {
323			// since we want to bubble the continue upstream
324			// we break here waiting for next $param if any
325			++$nodeStats['num_break'];
326			break;
327			}
328
329			// we drop one iteration
330			++$nodeStats['num_continue'];
331			continue;
332			}
333
334			if ($this->break) {
335			// we drop all subsequent iterations
336			++$nodeStats['num_break'];
337			$this->break = $this->interruptNode($node);
338			break;
339			}
340			}
341
342			// we reached the end of this Traversable and executed all its downstream Nodes
343			++$nodeStats['num_exec'];
344
345			return $param;
346			}
347
348			/** @var ExecNodeInterface $node */
349			$value = $node->exec($param);
350			++$nodeStats['num_exec'];
351
352			if ($this->continue) {
353			++$nodeStats['num_continue'];
354			// a continue does not need to bubble up unless
355			// it specifically targets a node in this flow
356			// or targets an upstream flow
357			$this->continue = $this->interruptNode($node);
358
359			return $param;
360			}
361
362			if ($this->break) {
363			++$nodeStats['num_break'];
364			// a break always need to bubble up to the first upstream Traversable if any
365			return $param;
366			}
367
368			if ($returnVal) {
369			// pass current $value as next param
370			$param = $value;
371			}
372
373			++$nodeIdx;
374			}
375
376			// we reached the end of this recursion
377			return $param;
378			}
379			}
380

fab2s / NodalFlow

Push — onFlowStart ( 676580...51dac1 )

NodalFlow::onFlowStart() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like