YaEtl::flushNodes() - Code Metrics - Inspection of "Prepare merge" - fab2s/YaEtl - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 7bba5d...c1c3fd )

by Fabrice

created 2017-07-27 18:40 UTC

YaEtl::flushNodes() B

↳ Parent: YaEtl

Complexity

Conditions	5
Paths	4

Size

Total Lines	18
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

<?php

/*
 * This file is part of YaEtl.
 *     (c) Fabrice de Stefanis / https://github.com/fab2s/YaEtl
 * 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\YaEtl;

use fab2s\NodalFlow\Flows\FlowStatusInterface;
use fab2s\NodalFlow\NodalFlow;
use fab2s\NodalFlow\Nodes\AggregateNode;
use fab2s\NodalFlow\Nodes\AggregateNodeInterface;
use fab2s\NodalFlow\Nodes\BranchNode;
use fab2s\NodalFlow\Nodes\NodeInterface;
use fab2s\NodalFlow\YaEtlException;
use fab2s\YaEtl\Extractors\ExtractorInterface;
use fab2s\YaEtl\Extractors\JoinableInterface;
use fab2s\YaEtl\Extractors\OnClauseInterface;
use fab2s\YaEtl\Loaders\LoaderInterface;
use fab2s\YaEtl\Transformers\TransformerInterface;

/**
 * Class YaEtl
 */
class YaEtl extends NodalFlow
{
    /**
     * The stats items added to NodalFlow's ones
     *
     * @var array
     */
    protected $stats = [
        'start'           => null,
        'end'             => null,
        'duration'        => null,
        'mib'             => null,
        'report'          => '',
        'num_extract'     => 0,
        'num_extractor'   => 0,
        'num_join'        => 0,
        'num_joiner'      => 0,
        'num_merge'       => 0,
        'num_records'     => 0,
        'num_transform'   => 0,
        'num_transformer' => 0,
        'num_branch'      => 0,
        'num_load'        => 0,
        'num_loader'      => 0,
        'num_flush'       => 0,
        'invocations'     => [],
        'nodes'           => [],
    ];

    /**
     * The revers aggregate lookup table
     *
     * @var array
     */
    protected $reverseAggregateTable = [];

    /**
     * @var bool
     */
    protected $forceFlush = false;

    /**
     * Adds an extractor to the Flow which may be aggregated with another one
     *
     * @param ExtractorInterface      $extractor
     * @param null|ExtractorInterface $aggregateWith Use the extractore instance you want to aggregate with
     *
     * @return $this
     */
    public function from(ExtractorInterface $extractor, ExtractorInterface $aggregateWith = null)
    {
        $this->enforceNodeInstanceUnicity($extractor);
        if ($aggregateWith !== null) {
            $this->aggregateTo($extractor, $aggregateWith);
        } else {
            parent::add($extractor);
        }

        ++$this->stats['num_extractor'];

        return $this;
    }

    /**
     * Override NodalFlow's add method to prohibit its direct usage
     *
     * @param NodeInterface $node
     *
     * @throws YaEtlException
     */
    public function add(NodeInterface $node)
    {
        throw new YaEtlException('add() is not directly available, use YaEtl grammar from(), transform(), join() and / or to() instead');
    }

    /**
     * By default, branched flows will only see their
     * `flush()` method called when the top most parent
     * triggers its own `flush()`.
     * It make sense most of the time to to do so as
     * the most common use case for branches so far is
     * to deal with one record at a time without generating
     * records (even when left joining). In such case,
     * the `flush()` method really need to be called by the flow
     * exactly when the top most flow one is.
     *
     * Set to true if you are generating many records in a branch
     * and it makes sense to flush the branch more often
     * Also note that the branch will also be flushed at the end
     * of its top most parent.
     *
     * @param bool $forceFlush
     *
     * @return $this
     */
    public function forceFlush($forceFlush)
    {
        $this->forceFlush = (bool) $forceFlush;

        return $this;
    }

    /**
     * Adds a Joiner to a specific Extractor in the FLow
     *
     * @param JoinableInterface $extractor
     * @param JoinableInterface $joinFrom
     * @param OnClauseInterface $onClause
     *
     * @return $this
     */
    public function join(JoinableInterface $extractor, JoinableInterface $joinFrom, OnClauseInterface $onClause)
    {
        $this->enforceNodeInstanceUnicity($extractor);
        $joinFrom->registerJoinerOnClause($onClause);
        $extractor->setJoinFrom($joinFrom);
        $extractor->setOnClause($onClause);

        parent::add($extractor);
        ++$this->stats['num_joiner'];

        return $this;
    }

    /**
     * Adds a Transformer to the Flow
     *
     * @param TransformerInterface $transformer
     *
     * @return $this
     */
    public function transform(TransformerInterface $transformer)
    {
        $this->enforceNodeInstanceUnicity($transformer);
        parent::add($transformer);
        ++$this->stats['num_transformer'];

        return $this;
    }

    /**
     * Adds a Loader to the Flow
     *
     * @param LoaderInterface $loader
     *
     * @return $this
     */
    public function to(LoaderInterface $loader)
    {
        $this->enforceNodeInstanceUnicity($loader);
        parent::add($loader);
        ++$this->stats['num_loader'];

        return $this;
    }

    /**
     * Adds a Branch (Flow) to the Flow
     *
     * @staticvar type $flowHashes
     *
     * @param YaEtl $flow            The Branch to add in this Flow
     * @param bool  $isAReturningVal To indicate if this Branch Flow is a true Branch or just
     *                               a bag of Nodes to execute at this location of the Flow
     *
     * @throws YaEtlException
     *
     * @return $this
     */
    public function branch(YaEtl $flow, $isAReturningVal = false)
    {
        static $flowHashes;
        if (!isset($flowHashes)) {
            $flowHashes = [
                $this->objectHash($this) => 1,
            ];
        }

        $flowHash = $this->objectHash($flow);
        if (isset($flowHashes[$flowHash])) {
            throw new YaEtlException('An instance of ' . \get_class($flow) . ' appears to be already in use in this flow. Please clone / re new before reuse');
        }

        $flowHashes[$flowHash] = 1;

        parent::add(new BranchNode($flow, $isAReturningVal));
        ++$this->stats['num_branch'];

        return $this;
    }

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

        parent::flowEnd();

        return $this;
    }

    /**
     * KISS method to expose basic stats
     *
     * @return array

     */
    public function getStats()
    {
        $stats          = $this->processStats(parent::getstats());
        $stats['nodes'] = $this->getNodeStats();

        $this->collectNodeStats($stats);

        $stats['duration'] = $stats['end'] - $stats['start'];
        $stats             = \array_replace($stats, $this->duration($stats['duration']));
        $stats['report']   = \sprintf(
            '[YaEtl](%s) %s Extractor - %s Extract - %s Record (%s Iterations)
[YaEtl] %s Joiner - %s Join - %s Continue - %s Break - %s Branch
[YaEtl] %s Transformer - %s Transform - %s Loader - %s Load - %s Flush
[YaEtl] Time : %s - Memory: %4.2fMiB',
            $this->flowStatus,
            \number_format($stats['num_extractor'], 0, '.', ' '),
            \number_format($stats['num_extract'], 0, '.', ' '),
            \number_format($stats['num_records'], 0, '.', ' '),
            \number_format($this->numIterate, 0, '.', ' '),
            \number_format($stats['num_joiner'], 0, '.', ' '),
            \number_format($stats['num_join'], 0, '.', ' '),
            \number_format($this->numContinue, 0, '.', ' '),
            \number_format($this->numBreak, 0, '.', ' '),
            \number_format($stats['num_branch'], 0, '.', ' '),
            \number_format($stats['num_transformer'], 0, '.', ' '),
            \number_format($stats['num_transform'], 0, '.', ' '),
            \number_format($stats['num_loader'], 0, '.', ' '),
            \number_format($stats['num_load'], 0, '.', ' '),
            \number_format($stats['num_flush'], 0, '.', ' '),
            $stats['durationStr'],
            $stats['mib']
        );

        return $stats;
    }

    /**
     * Tells if the flow is set to force flush
     * Only used when branched (to tell the parent)
     *
     * @return bool
     */
    public function isForceFlush()
    {
        return !empty($this->forceFlush);
    }

    /**
     * Used internally to aggregate Extracors
     *
     * @param ExtractorInterface $extractor
     * @param ExtractorInterface $aggregateWith
     *
     * @throws YaEtlException
     *
     * @return $this
     */
    protected function aggregateTo(ExtractorInterface $extractor, ExtractorInterface $aggregateWith)
    {
        // aggregate with target Node
        $nodeHash = $aggregateWith->getNodeHash();
        if (!isset($this->nodeMap[$nodeHash]) && !isset($this->reverseAggregateTable[$nodeHash])) {
            throw new YaEtlException('Cannot aggregate with orphaned Node:' . \get_class($aggregateWith));
        }

        $aggregateWithIdx = isset($this->nodeMap[$nodeHash]) ? $this->nodeMap[$nodeHash]['index'] : $this->reverseAggregateTable[$nodeHash];
        if ($this->nodes[$aggregateWithIdx] instanceof AggregateNodeInterface) {
            $this->nodes[$aggregateWithIdx]->addTraversable($extractor);
            // aggregate node did take care of setting carrier and hash
            $this->reverseAggregateTable[$extractor->getNodeHash()] = $aggregateWithIdx;

            return $this;
        }

        $aggregateNode = new AggregateNode(true);
        $aggregateNode->addTraversable($this->nodes[$aggregateWithIdx])
                ->addTraversable($extractor);
        // keep track of this extractor before we burry it in the aggregate
        $this->reverseAggregateTable[$this->nodes[$aggregateWithIdx]->getNodeHash()] = $aggregateWithIdx;
        // now replace its slot in the main tree
        $this->replace($aggregateWithIdx, $aggregateNode);
        // aggregate node did take care of setting carrier and hash
        $this->reverseAggregateTable[$aggregateNode->getNodeHash()]                  = $aggregateWithIdx;
        $this->reverseAggregateTable[$extractor->getNodeHash()]                      = $aggregateWithIdx;

        return $this;
    }

    /**
     * Collect Nodes stats
     *
     * @param array $stats
     *
     * @return $this
     */
    protected function collectNodeStats(array &$stats)
    {
        $stats = \array_replace($this->statsDefault, $stats);
        foreach ($this->nodes as $nodeIdx => $node) {
            if (($node instanceof JoinableInterface) && $node->getOnClause()) {
                $this->nodeStats[$nodeIdx]['num_join'] = $node->getNumRecords();
                $stats['num_join'] += $this->nodeStats[$nodeIdx]['num_join'];
            } elseif ($node instanceof ExtractorInterface) {
                $this->nodeStats[$nodeIdx]['num_records'] = $this->nodeStats[$nodeIdx]['num_iterate'];
                $this->nodeStats[$nodeIdx]['num_extract'] = $node->getNumExtract();
                $stats['num_records'] += $this->nodeStats[$nodeIdx]['num_iterate'];
                $stats['num_extract'] += $this->nodeStats[$nodeIdx]['num_extract'];
            } elseif ($node instanceof TransformerInterface) {
                $this->nodeStats[$nodeIdx]['num_transform'] = $this->nodeStats[$nodeIdx]['num_exec'];
                $stats['num_transform'] += $this->nodeStats[$nodeIdx]['num_transform'];
            } elseif ($node instanceof LoaderInterface) {
                $this->nodeStats[$nodeIdx]['num_load'] = $this->nodeStats[$nodeIdx]['num_exec'];
                $stats['num_load'] += $this->nodeStats[$nodeIdx]['num_load'];
            } elseif ($node instanceof AggregateNodeInterface) {
                $this->nodeStats[$nodeIdx]['num_records'] = $this->nodeStats[$nodeIdx]['num_iterate'];
                $stats['num_records'] += $this->nodeStats[$nodeIdx]['num_iterate'];
                $this->nodeStats[$nodeIdx]['num_extract'] = 0;
                foreach ($node->getNodeCollection() as $extractorNode) {
                    $this->nodeStats[$nodeIdx]['num_extract'] += $extractorNode->getNumExtract();
                }

                $stats['num_extract'] += $this->nodeStats[$nodeIdx]['num_extract'];
            }
        }

        return $this;
    }

    /**
     * Replaces a node with another one
     *
     * @param type          $nodeIdx
     * @param NodeInterface $node
     *
     * @throws YaEtlException
     *
     * @return $this
     */
    protected function replace($nodeIdx, NodeInterface $node)
    {
        if (!isset($this->nodes[$nodeIdx])) {
            throw new YaEtlException('Argument 1 should be a valid index in nodes, got:' . \gettype($nodeIdx));
        }

        unset($this->nodeMap[$this->nodeStats[$nodeIdx]['hash']], $this->nodeStats[$nodeIdx]);
        $nodeHash = $this->objectHash($node);

        $node->setCarrier($this)->setNodeHash($nodeHash);

        $this->nodes[$nodeIdx]    = $node;
        $this->nodeMap[$nodeHash] = \array_replace($this->nodeMapDefault, [
            'class'    => \get_class($node),
            'branchId' => $this->flowId,
            'hash'     => $nodeHash,
            'index'    => $nodeIdx,
        ]);

        // register references to nodeStats to increment faster
        // nodeStats can also be used as reverse lookup table
        $this->nodeStats[$nodeIdx] = &$this->nodeMap[$nodeHash];

        return $this;
    }

    /**
     * Compute final stats
     *
     * @param array $stats
     *
     * @return array
     */
    protected function processStats($stats)
    {
        if (!empty($stats['nodes'])) {
            $stats['nodes'] = $this->processStats($stats['nodes']);
        }

        if (empty($stats['invocations'])) {
            return $stats;
        }

        foreach ($stats['invocations'] as &$value) {
            $value           = \array_replace($value, $this->duration($value['duration']));
            $value['report'] = \sprintf('[YaEtl] Time : %s - Memory: %4.2fMiB',
                $value['durationStr'],
                $value['mib']
            );
        }

        return $stats;
    }

    /**
     * It could lead to really tricky situation if we where to
     * allow multiple instances of the same node. It's obviously
     * wrong with an Extractor, but even a Transformer could
     * create dark corner cases.
     *
     * @param NodeInterface $node
     *
     * @throws YaEtlException
     *
     * @return $this
     */
    protected function enforceNodeInstanceUnicity(NodeInterface $node)
    {
        if ($this->findNodeHashInMap($this->objectHash($node), $this->getNodeMap())) {
            throw new YaEtlException('This instance of ' . \get_class($node) . ' appears to be already in use in this flow. Please deep clone / re new before reuse');
        }

        return $this;
    }

    /**
     * Find a Node by its hash in a nodemap, used to enfore Node instance unicity
     *
     * @param string $hash
     * @param array  $nodeMap
     *
     * @return bool
     */
    protected function findNodeHashInMap($hash, $nodeMap)
    {
        if (isset($nodeMap[$hash])) {
            return true;
        }

        foreach ($nodeMap as $mapData) {
            if (
                !empty($mapData['nodes']) &&
                $this->findNodeHashInMap($hash, $mapData['nodes'])
            ) {
                return true;
            }
        }

        return false;
    }

    /**
     * Calls each WorkFlow's loaders and branch flush method
     *
     * @return $this
     */
    protected function flush(FlowStatusInterface $flowStatus = null)
    {
        if ($flowStatus === null) {
            if ($this->hasParent() && !$this->isForceFlush()) {
                // we'll get another chance at this
                return $this;
            }

            // use current status
            return $this->flushNodes($this->flowStatus);
        }

        // use parent's status
        return $this->flushNodes($flowStatus);
    }

    /**
     * Actually flush nodes
     *
     * @param FlowStatusInterface $flowStatus
     *
     * @return $this
     */
    protected function flushNodes(FlowStatusInterface $flowStatus)
    {
        foreach ($this->nodes as $node) {
            if ($node instanceof LoaderInterface) {
                $node->flush($flowStatus);
                ++$this->stats['num_flush'];
                continue;
            }

            // start with only flushing YaEtl and extends
            if ($node instanceof BranchNode && \is_a($node->getPayload(), static::class)) {
                $node->getPayload()->flush($flowStatus);

                ++$this->stats['num_flush'];
            }
        }

        return $this;
    }
}


1			<?php
2
3			/*
4			* This file is part of YaEtl.
5			* (c) Fabrice de Stefanis / https://github.com/fab2s/YaEtl
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\YaEtl;
11
12			use fab2s\NodalFlow\Flows\FlowStatusInterface;
13			use fab2s\NodalFlow\NodalFlow;
14			use fab2s\NodalFlow\Nodes\AggregateNode;
15			use fab2s\NodalFlow\Nodes\AggregateNodeInterface;
16			use fab2s\NodalFlow\Nodes\BranchNode;
17			use fab2s\NodalFlow\Nodes\NodeInterface;
18			use fab2s\NodalFlow\YaEtlException;
19			use fab2s\YaEtl\Extractors\ExtractorInterface;
20			use fab2s\YaEtl\Extractors\JoinableInterface;
21			use fab2s\YaEtl\Extractors\OnClauseInterface;
22			use fab2s\YaEtl\Loaders\LoaderInterface;
23			use fab2s\YaEtl\Transformers\TransformerInterface;
24
25			/**
26			* Class YaEtl
27			*/
28			class YaEtl extends NodalFlow
29			{
30			/**
31			* The stats items added to NodalFlow's ones
32			*
33			* @var array
34			*/
35			protected $stats = [
36			'start' => null,
37			'end' => null,
38			'duration' => null,
39			'mib' => null,
40			'report' => '',
41			'num_extract' => 0,
42			'num_extractor' => 0,
43			'num_join' => 0,
44			'num_joiner' => 0,
45			'num_merge' => 0,
46			'num_records' => 0,
47			'num_transform' => 0,
48			'num_transformer' => 0,
49			'num_branch' => 0,
50			'num_load' => 0,
51			'num_loader' => 0,
52			'num_flush' => 0,
53			'invocations' => [],
54			'nodes' => [],
55			];
56
57			/**
58			* The revers aggregate lookup table
59			*
60			* @var array
61			*/
62			protected $reverseAggregateTable = [];
63
64			/**
65			* @var bool
66			*/
67			protected $forceFlush = false;
68
69			/**
70			* Adds an extractor to the Flow which may be aggregated with another one
71			*
72			* @param ExtractorInterface $extractor
73			* @param null\|ExtractorInterface $aggregateWith Use the extractore instance you want to aggregate with
74			*
75			* @return $this
76			*/
77			public function from(ExtractorInterface $extractor, ExtractorInterface $aggregateWith = null)
78			{
79			$this->enforceNodeInstanceUnicity($extractor);
80			if ($aggregateWith !== null) {
81			$this->aggregateTo($extractor, $aggregateWith);
82			} else {
83			parent::add($extractor);
84			}
85
86			++$this->stats['num_extractor'];
87
88			return $this;
89			}
90
91			/**
92			* Override NodalFlow's add method to prohibit its direct usage
93			*
94			* @param NodeInterface $node
95			*
96			* @throws YaEtlException
97			*/
98			public function add(NodeInterface $node)
99			{
100			throw new YaEtlException('add() is not directly available, use YaEtl grammar from(), transform(), join() and / or to() instead');
101			}
102
103			/**
104			* By default, branched flows will only see their
105			* `flush()` method called when the top most parent
106			* triggers its own `flush()`.
107			* It make sense most of the time to to do so as
108			* the most common use case for branches so far is
109			* to deal with one record at a time without generating
110			* records (even when left joining). In such case,
111			* the `flush()` method really need to be called by the flow
112			* exactly when the top most flow one is.
113			*
114			* Set to true if you are generating many records in a branch
115			* and it makes sense to flush the branch more often
116			* Also note that the branch will also be flushed at the end
117			* of its top most parent.
118			*
119			* @param bool $forceFlush
120			*
121			* @return $this
122			*/
123			public function forceFlush($forceFlush)
124			{
125			$this->forceFlush = (bool) $forceFlush;
126
127			return $this;
128			}
129
130			/**
131			* Adds a Joiner to a specific Extractor in the FLow
132			*
133			* @param JoinableInterface $extractor
134			* @param JoinableInterface $joinFrom
135			* @param OnClauseInterface $onClause
136			*
137			* @return $this
138			*/
139			public function join(JoinableInterface $extractor, JoinableInterface $joinFrom, OnClauseInterface $onClause)
140			{
141			$this->enforceNodeInstanceUnicity($extractor);
142			$joinFrom->registerJoinerOnClause($onClause);
143			$extractor->setJoinFrom($joinFrom);
144			$extractor->setOnClause($onClause);
145
146			parent::add($extractor);
147			++$this->stats['num_joiner'];
148
149			return $this;
150			}
151
152			/**
153			* Adds a Transformer to the Flow
154			*
155			* @param TransformerInterface $transformer
156			*
157			* @return $this
158			*/
159			public function transform(TransformerInterface $transformer)
160			{
161			$this->enforceNodeInstanceUnicity($transformer);
162			parent::add($transformer);
163			++$this->stats['num_transformer'];
164
165			return $this;
166			}
167
168			/**
169			* Adds a Loader to the Flow
170			*
171			* @param LoaderInterface $loader
172			*
173			* @return $this
174			*/
175			public function to(LoaderInterface $loader)
176			{
177			$this->enforceNodeInstanceUnicity($loader);
178			parent::add($loader);
179			++$this->stats['num_loader'];
180
181			return $this;
182			}
183
184			/**
185			* Adds a Branch (Flow) to the Flow
186			*
187			* @staticvar type $flowHashes
188			*
189			* @param YaEtl $flow The Branch to add in this Flow
190			* @param bool $isAReturningVal To indicate if this Branch Flow is a true Branch or just
191			* a bag of Nodes to execute at this location of the Flow
192			*
193			* @throws YaEtlException
194			*
195			* @return $this
196			*/
197			public function branch(YaEtl $flow, $isAReturningVal = false)
198			{
199			static $flowHashes;
200			if (!isset($flowHashes)) {
201			$flowHashes = [
202			$this->objectHash($this) => 1,
203			];
204			}
205
206			$flowHash = $this->objectHash($flow);
207			if (isset($flowHashes[$flowHash])) {
208			throw new YaEtlException('An instance of ' . \get_class($flow) . ' appears to be already in use in this flow. Please clone / re new before reuse');
209			}
210
211			$flowHashes[$flowHash] = 1;
212
213			parent::add(new BranchNode($flow, $isAReturningVal));
214			++$this->stats['num_branch'];
215
216			return $this;
217			}
218
219			/**
220			* Triggered right after the flow stops
221			*
222			* @return $this
223			*/
224			public function flowEnd()
225			{
226			$this->flush();
227
228			parent::flowEnd();
229
230			return $this;
231			}
232
233			/**
234			* KISS method to expose basic stats
235			*
236			* @return array
			0 ignored issues – show Documentation introduced 2017-05-04 18:16 UTC by Report Bug Copy Issue Report Consider making the return type a bit more specific; maybe use `array<string,integer\|double\|string>`. This check looks for the generic type `array` as a return type and suggests a more specific type. This type is inferred from the actual code. Loading history...
237			*/
238			public function getStats()
239			{
240			$stats = $this->processStats(parent::getstats());
241			$stats['nodes'] = $this->getNodeStats();
242
243			$this->collectNodeStats($stats);
244
245			$stats['duration'] = $stats['end'] - $stats['start'];
246			$stats = \array_replace($stats, $this->duration($stats['duration']));
247			$stats['report'] = \sprintf(
248			'[YaEtl](%s) %s Extractor - %s Extract - %s Record (%s Iterations)
249			[YaEtl] %s Joiner - %s Join - %s Continue - %s Break - %s Branch
250			[YaEtl] %s Transformer - %s Transform - %s Loader - %s Load - %s Flush
251			[YaEtl] Time : %s - Memory: %4.2fMiB',
252			$this->flowStatus,
253			\number_format($stats['num_extractor'], 0, '.', ' '),
254			\number_format($stats['num_extract'], 0, '.', ' '),
255			\number_format($stats['num_records'], 0, '.', ' '),
256			\number_format($this->numIterate, 0, '.', ' '),
257			\number_format($stats['num_joiner'], 0, '.', ' '),
258			\number_format($stats['num_join'], 0, '.', ' '),
259			\number_format($this->numContinue, 0, '.', ' '),
260			\number_format($this->numBreak, 0, '.', ' '),
261			\number_format($stats['num_branch'], 0, '.', ' '),
262			\number_format($stats['num_transformer'], 0, '.', ' '),
263			\number_format($stats['num_transform'], 0, '.', ' '),
264			\number_format($stats['num_loader'], 0, '.', ' '),
265			\number_format($stats['num_load'], 0, '.', ' '),
266			\number_format($stats['num_flush'], 0, '.', ' '),
267			$stats['durationStr'],
268			$stats['mib']
269			);
270
271			return $stats;
272			}
273
274			/**
275			* Tells if the flow is set to force flush
276			* Only used when branched (to tell the parent)
277			*
278			* @return bool
279			*/
280			public function isForceFlush()
281			{
282			return !empty($this->forceFlush);
283			}
284
285			/**
286			* Used internally to aggregate Extracors
287			*
288			* @param ExtractorInterface $extractor
289			* @param ExtractorInterface $aggregateWith
290			*
291			* @throws YaEtlException
292			*
293			* @return $this
294			*/
295			protected function aggregateTo(ExtractorInterface $extractor, ExtractorInterface $aggregateWith)
296			{
297			// aggregate with target Node
298			$nodeHash = $aggregateWith->getNodeHash();
299			if (!isset($this->nodeMap[$nodeHash]) && !isset($this->reverseAggregateTable[$nodeHash])) {
300			throw new YaEtlException('Cannot aggregate with orphaned Node:' . \get_class($aggregateWith));
301			}
302
303			$aggregateWithIdx = isset($this->nodeMap[$nodeHash]) ? $this->nodeMap[$nodeHash]['index'] : $this->reverseAggregateTable[$nodeHash];
304			if ($this->nodes[$aggregateWithIdx] instanceof AggregateNodeInterface) {
305			$this->nodes[$aggregateWithIdx]->addTraversable($extractor);
306			// aggregate node did take care of setting carrier and hash
307			$this->reverseAggregateTable[$extractor->getNodeHash()] = $aggregateWithIdx;
308
309			return $this;
310			}
311
312			$aggregateNode = new AggregateNode(true);
313			$aggregateNode->addTraversable($this->nodes[$aggregateWithIdx])
314			->addTraversable($extractor);
315			// keep track of this extractor before we burry it in the aggregate
316			$this->reverseAggregateTable[$this->nodes[$aggregateWithIdx]->getNodeHash()] = $aggregateWithIdx;
317			// now replace its slot in the main tree
318			$this->replace($aggregateWithIdx, $aggregateNode);
319			// aggregate node did take care of setting carrier and hash
320			$this->reverseAggregateTable[$aggregateNode->getNodeHash()] = $aggregateWithIdx;
321			$this->reverseAggregateTable[$extractor->getNodeHash()] = $aggregateWithIdx;
322
323			return $this;
324			}
325
326			/**
327			* Collect Nodes stats
328			*
329			* @param array $stats
330			*
331			* @return $this
332			*/
333			protected function collectNodeStats(array &$stats)
334			{
335			$stats = \array_replace($this->statsDefault, $stats);
336			foreach ($this->nodes as $nodeIdx => $node) {
337			if (($node instanceof JoinableInterface) && $node->getOnClause()) {
338			$this->nodeStats[$nodeIdx]['num_join'] = $node->getNumRecords();
339			$stats['num_join'] += $this->nodeStats[$nodeIdx]['num_join'];
340			} elseif ($node instanceof ExtractorInterface) {
341			$this->nodeStats[$nodeIdx]['num_records'] = $this->nodeStats[$nodeIdx]['num_iterate'];
342			$this->nodeStats[$nodeIdx]['num_extract'] = $node->getNumExtract();
343			$stats['num_records'] += $this->nodeStats[$nodeIdx]['num_iterate'];
344			$stats['num_extract'] += $this->nodeStats[$nodeIdx]['num_extract'];
345			} elseif ($node instanceof TransformerInterface) {
346			$this->nodeStats[$nodeIdx]['num_transform'] = $this->nodeStats[$nodeIdx]['num_exec'];
347			$stats['num_transform'] += $this->nodeStats[$nodeIdx]['num_transform'];
348			} elseif ($node instanceof LoaderInterface) {
349			$this->nodeStats[$nodeIdx]['num_load'] = $this->nodeStats[$nodeIdx]['num_exec'];
350			$stats['num_load'] += $this->nodeStats[$nodeIdx]['num_load'];
351			} elseif ($node instanceof AggregateNodeInterface) {
352			$this->nodeStats[$nodeIdx]['num_records'] = $this->nodeStats[$nodeIdx]['num_iterate'];
353			$stats['num_records'] += $this->nodeStats[$nodeIdx]['num_iterate'];
354			$this->nodeStats[$nodeIdx]['num_extract'] = 0;
355			foreach ($node->getNodeCollection() as $extractorNode) {
356			$this->nodeStats[$nodeIdx]['num_extract'] += $extractorNode->getNumExtract();
357			}
358
359			$stats['num_extract'] += $this->nodeStats[$nodeIdx]['num_extract'];
360			}
361			}
362
363			return $this;
364			}
365
366			/**
367			* Replaces a node with another one
368			*
369			* @param type $nodeIdx
370			* @param NodeInterface $node
371			*
372			* @throws YaEtlException
373			*
374			* @return $this
375			*/
376			protected function replace($nodeIdx, NodeInterface $node)
377			{
378			if (!isset($this->nodes[$nodeIdx])) {
379			throw new YaEtlException('Argument 1 should be a valid index in nodes, got:' . \gettype($nodeIdx));
380			}
381
382			unset($this->nodeMap[$this->nodeStats[$nodeIdx]['hash']], $this->nodeStats[$nodeIdx]);
383			$nodeHash = $this->objectHash($node);
384
385			$node->setCarrier($this)->setNodeHash($nodeHash);
386
387			$this->nodes[$nodeIdx] = $node;
388			$this->nodeMap[$nodeHash] = \array_replace($this->nodeMapDefault, [
389			'class' => \get_class($node),
390			'branchId' => $this->flowId,
391			'hash' => $nodeHash,
392			'index' => $nodeIdx,
393			]);
394
395			// register references to nodeStats to increment faster
396			// nodeStats can also be used as reverse lookup table
397			$this->nodeStats[$nodeIdx] = &$this->nodeMap[$nodeHash];
398
399			return $this;
400			}
401
402			/**
403			* Compute final stats
404			*
405			* @param array $stats
406			*
407			* @return array
408			*/
409			protected function processStats($stats)
410			{
411			if (!empty($stats['nodes'])) {
412			$stats['nodes'] = $this->processStats($stats['nodes']);
413			}
414
415			if (empty($stats['invocations'])) {
416			return $stats;
417			}
418
419			foreach ($stats['invocations'] as &$value) {
420			$value = \array_replace($value, $this->duration($value['duration']));
421			$value['report'] = \sprintf('[YaEtl] Time : %s - Memory: %4.2fMiB',
422			$value['durationStr'],
423			$value['mib']
424			);
425			}
426
427			return $stats;
428			}
429
430			/**
431			* It could lead to really tricky situation if we where to
432			* allow multiple instances of the same node. It's obviously
433			* wrong with an Extractor, but even a Transformer could
434			* create dark corner cases.
435			*
436			* @param NodeInterface $node
437			*
438			* @throws YaEtlException
439			*
440			* @return $this
441			*/
442			protected function enforceNodeInstanceUnicity(NodeInterface $node)
443			{
444			if ($this->findNodeHashInMap($this->objectHash($node), $this->getNodeMap())) {
445			throw new YaEtlException('This instance of ' . \get_class($node) . ' appears to be already in use in this flow. Please deep clone / re new before reuse');
446			}
447
448			return $this;
449			}
450
451			/**
452			* Find a Node by its hash in a nodemap, used to enfore Node instance unicity
453			*
454			* @param string $hash
455			* @param array $nodeMap
456			*
457			* @return bool
458			*/
459			protected function findNodeHashInMap($hash, $nodeMap)
460			{
461			if (isset($nodeMap[$hash])) {
462			return true;
463			}
464
465			foreach ($nodeMap as $mapData) {
466			if (
467			!empty($mapData['nodes']) &&
468			$this->findNodeHashInMap($hash, $mapData['nodes'])
469			) {
470			return true;
471			}
472			}
473
474			return false;
475			}
476
477			/**
478			* Calls each WorkFlow's loaders and branch flush method
479			*
480			* @return $this
481			*/
482			protected function flush(FlowStatusInterface $flowStatus = null)
483			{
484			if ($flowStatus === null) {
485			if ($this->hasParent() && !$this->isForceFlush()) {
486			// we'll get another chance at this
487			return $this;
488			}
489
490			// use current status
491			return $this->flushNodes($this->flowStatus);
492			}
493
494			// use parent's status
495			return $this->flushNodes($flowStatus);
496			}
497
498			/**
499			* Actually flush nodes
500			*
501			* @param FlowStatusInterface $flowStatus
502			*
503			* @return $this
504			*/
505			protected function flushNodes(FlowStatusInterface $flowStatus)
506			{
507			foreach ($this->nodes as $node) {
508			if ($node instanceof LoaderInterface) {
509			$node->flush($flowStatus);
510			++$this->stats['num_flush'];
511			continue;
512			}
513
514			// start with only flushing YaEtl and extends
515			if ($node instanceof BranchNode && \is_a($node->getPayload(), static::class)) {
516			$node->getPayload()->flush($flowStatus);
			0 ignored issues – show Bug introduced 2017-07-27 18:42 UTC by Report Bug Copy Issue Report The method `flush` cannot be called on `$node->getPayload()` (of type `callable`). Methods can only be called on objects. This check looks for methods being called on variables that have been inferred to never be objects. Loading history...
517			++$this->stats['num_flush'];
518			}
519			}
520
521			return $this;
522			}
523			}
524

fab2s / YaEtl

Push — master ( 7bba5d...c1c3fd )

YaEtl::flushNodes() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like