StreamQueryResult::all() - Code Metrics - Inspection of "Add StreamQueryResult::all and QueryResult::getRow..." - PlasmaPHP/core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 445a46...cd0fa9 )

by Charlotte

created 2018-11-28 17:33 UTC

StreamQueryResult::all() A

↳ Parent: StreamQueryResult

Complexity

Conditions	1
Paths	1

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nc	1
nop	0
dl	0
loc	3
ccs	2
cts	2
cp	1
crap	1
rs	10
c	0
b	0
f	0

<?php
/**
 * Plasma Core component
 * Copyright 2018 PlasmaPHP, All Rights Reserved
 *
 * Website: https://github.com/PlasmaPHP
 * License: https://github.com/PlasmaPHP/core/blob/master/LICENSE
*/

namespace Plasma;

/**
 * A query result stream. Used to get rows row by row, as sent by the DBMS.
 */
class StreamQueryResult implements StreamQueryResultInterface {
    use \Evenement\EventEmitterTrait;
    
    /**
     * @var \Plasma\DriverInterface
     */
    protected $driver;
    
    /**
     * @var \Plasma\CommandInterface
     */
    protected $command;
    
    /**
     * @var int
     */
    protected $affectedRows;
    
    /**
     * @var int
     */
    protected $warningsCount;
    
    /**
     * @var int|null
     */
    protected $insertID;
    
    /**
     * @var \Plasma\ColumnDefinitionInterface[]|null
     */
    protected $columns;
    
    /**
     * @var array|null
     */
    protected $rows;
    
    /**
     * @var bool
     */
    protected $started = false;
    
    /**
     * @var bool
     */
    protected $closed = false;
    
    /**
     * @var bool
     */
    protected $paused = false;
    
    /**
     * Constructor.
     * @param \Plasma\DriverInterface                   $driver
     * @param \Plasma\CommandInterface                  $command
     * @param int                                       $affectedRows
     * @param int                                       $warningsCount
     * @param int|null                                  $insertID
     * @param \Plasma\ColumnDefinitionInterface[]|null  $columns
     */
    function __construct(\Plasma\DriverInterface $driver, \Plasma\CommandInterface $command, int $affectedRows = 0, int $warningsCount = 0, ?int $insertID = null, ?array $columns = null) {
        $this->driver = $driver;
        $this->command = $command;
        
        $this->affectedRows = $affectedRows;
        $this->warningsCount = $warningsCount;
        
        $this->insertID = $insertID;
        $this->columns = $columns;
        
        $command->on('data', function ($row) {
            if(!$this->started && $this->paused) {
                $this->driver->pauseStreamConsumption();
            }
            
            $this->started = true;
            $this->emit('data', array($row));
        });
        
        $command->on('end', function () {
            $this->emit('end');
            $this->close();
        });
        
        $command->on('error', function (\Throwable $error) {
            $this->emit('error', array($error));
            $this->close();
        });
    }
    
    /**
     * Get the number of affected rows (for UPDATE, DELETE, etc.).
     * @return int
     */
    function getAffectedRows(): int {
        return $this->affectedRows;
    }
    
    /**
     * Get the number of warnings sent by the server.
     * @return int
     */
    function getWarningsCount(): int {
        return $this->warningsCount;
    }
    
    /**
     * Get the used insert ID for the row, if any. `INSERT` statements only.
     * @return int|null
     */
    function getInsertID(): ?int {
        return $this->insertID;
    }
    
    /**
     * Get the field definitions, if any. `SELECT` statements only.
     * @return \Plasma\ColumnDefinitionInterface[]|null
     */
    function getFieldDefinitions(): ?array {
        return $this->columns;
    }
    
    /**
     * Get the rows, if any. Returns always `null`.
     * @return array|null
     */
    function getRows(): ?array {
        return null;
    }
    
    /**
     * Buffers all rows and returns a promise which resolves with an instance of `QueryResultInterface`.
     * @return \React\Promise\PromiseInterface
     */
    function all(): \React\Promise\PromiseInterface {
        return \React\Promise\Stream\all($this)->then(function (array $rows) {
            return (new \Plasma\QueryResult($this->affectedRows, $this->warningsCount, $this->insertID, $this->columns, $rows));
        });
    }
    
    /**
     * Whether the stream is readable.
     * @return bool
     */
    function isReadable() {
        return (!$this->closed);
    }
    
    /**
     * Pauses the connection, where this stream is coming from.
     * This operation halts ALL read activities. You may still receive
     * `data` events until the underlying network buffer is drained.
     * @return void
     */
    function pause() {
        $this->paused = true;
        
        if($this->started && !$this->closed) {
            $this->driver->pauseStreamConsumption();
        }
    }
    
    /**
     * Resumes the connection, where this stream is coming from.
     * @return void
     */
    function resume() {
        $this->paused = false;
        
        if($this->started && !$this->closed) {
            $this->driver->resumeStreamConsumption();
        }
    }
    
    /**
     * Closes the stream. Resumes the connection stream.
     * @return void
     */
    function close() {
        if($this->closed) {
            return;
        }
        
        $this->closed = true;
        if($this->started && $this->paused) {
            $this->driver->resumeStreamConsumption();
        }
        
        $this->emit('close');
        $this->removeAllListeners();
    }
    
    /**
     * Pipes all the data from this readable source into the given writable destination.
     * Automatically sends all incoming data to the destination.
     * Automatically throttles the source based on what the destination can handle.
     * @param \React\Stream\WritableStreamInterface  $dest
     * @param array                                  $options
     * @return \React\Stream\WritableStreamInterface  $dest  Stream as-is
     */
    function pipe(\React\Stream\WritableStreamInterface $dest, array $options = array()) {
        return \React\Stream\Util::pipe($this, $dest, $options);
    }
}


1		<?php
2		/**
3		* Plasma Core component
4		* Copyright 2018 PlasmaPHP, All Rights Reserved
5		*
6		* Website: https://github.com/PlasmaPHP
7		* License: https://github.com/PlasmaPHP/core/blob/master/LICENSE
8		*/
9
10		namespace Plasma;
11
12		/**
13		* A query result stream. Used to get rows row by row, as sent by the DBMS.
14		*/
15		class StreamQueryResult implements StreamQueryResultInterface {
16		use \Evenement\EventEmitterTrait;
17
18		/**
19		* @var \Plasma\DriverInterface
20		*/
21		protected $driver;
22
23		/**
24		* @var \Plasma\CommandInterface
25		*/
26		protected $command;
27
28		/**
29		* @var int
30		*/
31		protected $affectedRows;
32
33		/**
34		* @var int
35		*/
36		protected $warningsCount;
37
38		/**
39		* @var int\|null
40		*/
41		protected $insertID;
42
43		/**
44		* @var \Plasma\ColumnDefinitionInterface[]\|null
45		*/
46		protected $columns;
47
48		/**
49		* @var array\|null
50		*/
51		protected $rows;
52
53		/**
54		* @var bool
55		*/
56		protected $started = false;
57
58		/**
59		* @var bool
60		*/
61		protected $closed = false;
62
63		/**
64		* @var bool
65		*/
66		protected $paused = false;
67
68		/**
69		* Constructor.
70		* @param \Plasma\DriverInterface $driver
71		* @param \Plasma\CommandInterface $command
72		* @param int $affectedRows
73		* @param int $warningsCount
74		* @param int\|null $insertID
75		* @param \Plasma\ColumnDefinitionInterface[]\|null $columns
76		*/
77	11	function __construct(\Plasma\DriverInterface $driver, \Plasma\CommandInterface $command, int $affectedRows = 0, int $warningsCount = 0, ?int $insertID = null, ?array $columns = null) {
78	11	$this->driver = $driver;
79	11	$this->command = $command;
80
81	11	$this->affectedRows = $affectedRows;
82	11	$this->warningsCount = $warningsCount;
83
84	11	$this->insertID = $insertID;
85	11	$this->columns = $columns;
86
87		$command->on('data', function ($row) {
88	2	if(!$this->started && $this->paused) {
89		$this->driver->pauseStreamConsumption();
90		}
91
92	2	$this->started = true;
93	2	$this->emit('data', array($row));
94	11	});
95
96		$command->on('end', function () {
97	1	$this->emit('end');
98	1	$this->close();
99	11	});
100
101		$command->on('error', function (\Throwable $error) {
102	1	$this->emit('error', array($error));
103	1	$this->close();
104	11	});
105	11	}
106
107		/**
108		* Get the number of affected rows (for UPDATE, DELETE, etc.).
109		* @return int
110		*/
111	1	function getAffectedRows(): int {
112	1	return $this->affectedRows;
113		}
114
115		/**
116		* Get the number of warnings sent by the server.
117		* @return int
118		*/
119	1	function getWarningsCount(): int {
120	1	return $this->warningsCount;
121		}
122
123		/**
124		* Get the used insert ID for the row, if any. `INSERT` statements only.
125		* @return int\|null
126		*/
127	1	function getInsertID(): ?int {
128	1	return $this->insertID;
129		}
130
131		/**
132		* Get the field definitions, if any. `SELECT` statements only.
133		* @return \Plasma\ColumnDefinitionInterface[]\|null
134		*/
135	1	function getFieldDefinitions(): ?array {
136	1	return $this->columns;
137		}
138
139		/**
140		* Get the rows, if any. Returns always `null`.
141		* @return array\|null
142		*/
143	1	function getRows(): ?array {
144	1	return null;
145		}
146
147		/**
148		* Buffers all rows and returns a promise which resolves with an instance of `QueryResultInterface`.
149		* @return \React\Promise\PromiseInterface
150		*/
151	1	function all(): \React\Promise\PromiseInterface {
152		return \React\Promise\Stream\all($this)->then(function (array $rows) {
153	1	return (new \Plasma\QueryResult($this->affectedRows, $this->warningsCount, $this->insertID, $this->columns, $rows));
154	1	});
155		}
156
157		/**
158		* Whether the stream is readable.
159		* @return bool
160		*/
161	4	function isReadable() {
162	4	return (!$this->closed);
163		}
164
165		/**
166		* Pauses the connection, where this stream is coming from.
167		* This operation halts ALL read activities. You may still receive
168		* `data` events until the underlying network buffer is drained.
169		* @return void
170		*/
171	2	function pause() {
172	2	$this->paused = true;
173
174	2	if($this->started && !$this->closed) {
175	2	$this->driver->pauseStreamConsumption();
176		}
177	2	}
178
179		/**
180		* Resumes the connection, where this stream is coming from.
181		* @return void
182		*/
183	1	function resume() {
184	1	$this->paused = false;
185
186	1	if($this->started && !$this->closed) {
187	1	$this->driver->resumeStreamConsumption();
188		}
189	1	}
190
191		/**
192		* Closes the stream. Resumes the connection stream.
193		* @return void
194		*/
195	3	function close() {
196	3	if($this->closed) {
197	1	return;
198		}
199
200	3	$this->closed = true;
201	3	if($this->started && $this->paused) {
202		$this->driver->resumeStreamConsumption();
203		}
204
205	3	$this->emit('close');
206	3	$this->removeAllListeners();
207	3	}
208
209		/**
210		* Pipes all the data from this readable source into the given writable destination.
211		* Automatically sends all incoming data to the destination.
212		* Automatically throttles the source based on what the destination can handle.
213		* @param \React\Stream\WritableStreamInterface $dest
214		* @param array $options
215		* @return \React\Stream\WritableStreamInterface $dest Stream as-is
216		*/
217		function pipe(\React\Stream\WritableStreamInterface $dest, array $options = array()) {
218		return \React\Stream\Util::pipe($this, $dest, $options);
219		}
220		}
221

PlasmaPHP / core

GitHub Access Token became invalid

Push — master ( 445a46...cd0fa9 )

StreamQueryResult::all() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like