StreamQueryResult::__construct() - Code Metrics - Inspection of "Stream Query Result tests" - PlasmaPHP/core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 659573...63ab33 )

by Charlotte

created 2018-11-12 16:35 UTC

StreamQueryResult::__construct() A

↳ Parent: StreamQueryResult

Complexity

Conditions	3
Paths	1

Size

Total Lines	27
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	16
CRAP Score	3.0017

Importance

Changes

Metric	Value
cc	3
eloc	17
nc	1
nop	6
dl	0
loc	27
ccs	16
cts	17
cp	0.9412
crap	3.0017
rs	9.7
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 = 0;
    
    /**
     * @var int
     */
    protected $warningsCount = 0;
    
    /**
     * @var \Plasma\ColumnDefinitionInterface[]|null
     */
    protected $fields;
    
    /**
     * @var int|null
     */
    protected $insertID = 0;
    
    /**
     * @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 \Plasma\ColumnDefinitionInterface[]|null  $fields
     * @param int|null                                  $insertID
     */
    function __construct(\Plasma\DriverInterface $driver, \Plasma\CommandInterface $command, int $affectedRows = 0, int $warningsCount = 0, ?array $fields = null, ?int $insertID = null) {
        $this->driver = $driver;
        $this->command = $command;
        
        $this->affectedRows = $affectedRows;
        $this->warningsCount = $warningsCount;
        
        $this->fields = $fields;
        $this->insertID = $insertID;
        
        $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 field definitions, if any. `SELECT` statements only.
     * @return \Plasma\ColumnDefinitionInterface[]|null
     */
    function getFieldDefinitions(): ?array {
        return $this->fields;
    }
    
    /**
     * Get the used insert ID for the row, if any. `INSERT` statements only.
     * @return int|null
     */
    function getInsertID(): ?int {
        return $this->insertID;
    }
    
    /**
     * 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 = 0;
32
33		/**
34		* @var int
35		*/
36		protected $warningsCount = 0;
37
38		/**
39		* @var \Plasma\ColumnDefinitionInterface[]\|null
40		*/
41		protected $fields;
42
43		/**
44		* @var int\|null
45		*/
46		protected $insertID = 0;
47
48		/**
49		* @var bool
50		*/
51		protected $started = false;
52
53		/**
54		* @var bool
55		*/
56		protected $closed = false;
57
58		/**
59		* @var bool
60		*/
61		protected $paused = false;
62
63		/**
64		* Constructor.
65		* @param \Plasma\DriverInterface $driver
66		* @param \Plasma\CommandInterface $command
67		* @param int $affectedRows
68		* @param int $warningsCount
69		* @param \Plasma\ColumnDefinitionInterface[]\|null $fields
70		* @param int\|null $insertID
71		*/
72	9	function __construct(\Plasma\DriverInterface $driver, \Plasma\CommandInterface $command, int $affectedRows = 0, int $warningsCount = 0, ?array $fields = null, ?int $insertID = null) {
73	9	$this->driver = $driver;
74	9	$this->command = $command;
75
76	9	$this->affectedRows = $affectedRows;
77	9	$this->warningsCount = $warningsCount;
78
79	9	$this->fields = $fields;
80	9	$this->insertID = $insertID;
81
82		$command->on('data', function ($row) {
83	2	if(!$this->started && $this->paused) {
84		$this->driver->pauseStreamConsumption();
85		}
86
87	2	$this->started = true;
88	2	$this->emit('data', array($row));
89	9	});
90
91		$command->on('end', function () {
92	1	$this->emit('end');
93	1	$this->close();
94	9	});
95
96		$command->on('error', function (\Throwable $error) {
97	1	$this->emit('error', array($error));
98	1	$this->close();
99	9	});
100	9	}
101
102		/**
103		* Get the number of affected rows (for UPDATE, DELETE, etc.).
104		* @return int
105		*/
106	1	function getAffectedRows(): int {
107	1	return $this->affectedRows;
108		}
109
110		/**
111		* Get the number of warnings sent by the server.
112		* @return int
113		*/
114	1	function getWarningsCount(): int {
115	1	return $this->warningsCount;
116		}
117
118		/**
119		* Get the field definitions, if any. `SELECT` statements only.
120		* @return \Plasma\ColumnDefinitionInterface[]\|null
121		*/
122	1	function getFieldDefinitions(): ?array {
123	1	return $this->fields;
124		}
125
126		/**
127		* Get the used insert ID for the row, if any. `INSERT` statements only.
128		* @return int\|null
129		*/
130	1	function getInsertID(): ?int {
131	1	return $this->insertID;
132		}
133
134		/**
135		* Whether the stream is readable.
136		* @return bool
137		*/
138	3	function isReadable() {
139	3	return (!$this->closed);
140		}
141
142		/**
143		* Pauses the connection, where this stream is coming from.
144		* This operation halts ALL read activities. You may still receive
145		* `data` events until the underlying network buffer is drained.
146		* @return void
147		*/
148	2	function pause() {
149	2	$this->paused = true;
150
151	2	if($this->started && !$this->closed) {
152	2	$this->driver->pauseStreamConsumption();
153		}
154	2	}
155
156		/**
157		* Resumes the connection, where this stream is coming from.
158		* @return void
159		*/
160	1	function resume() {
161	1	$this->paused = false;
162
163	1	if($this->started && !$this->closed) {
164	1	$this->driver->resumeStreamConsumption();
165		}
166	1	}
167
168		/**
169		* Closes the stream. Resumes the connection stream.
170		* @return void
171		*/
172	3	function close() {
173	3	if($this->closed) {
174		return;
175		}
176
177	3	$this->closed = true;
178	3	if($this->started && $this->paused) {
179		$this->driver->resumeStreamConsumption();
180		}
181
182	3	$this->emit('close');
183	3	$this->removeAllListeners();
184	3	}
185
186		/**
187		* Pipes all the data from this readable source into the given writable destination.
188		* Automatically sends all incoming data to the destination.
189		* Automatically throttles the source based on what the destination can handle.
190		* @param \React\Stream\WritableStreamInterface $dest
191		* @param array $options
192		* @return \React\Stream\WritableStreamInterface $dest Stream as-is
193		*/
194		function pipe(\React\Stream\WritableStreamInterface $dest, array $options = array()) {
195		return \React\Stream\Util::pipe($this, $dest, $options);
196		}
197		}
198

PlasmaPHP / core

GitHub Access Token became invalid

Push — master ( 659573...63ab33 )

StreamQueryResult::__construct() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like