ProtocolParser::addEvents() - Code Metrics - Inspection of "Implement #16" - PlasmaPHP/driver-mysql - Measure and Improve Code Quality continuously with Scrutinizer

Failed Conditions

Push — master ( 51f19d...b0fd42 )

by Charlotte

created 2019-03-18 19:37 UTC

ProtocolParser::addEvents() A

↳ Parent: ProtocolParser

Complexity

Conditions	4
Paths	1

Size

Total Lines	17
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	9
CRAP Score	4

Importance

Changes

Metric	Value
cc	4
eloc	10
nc	1
nop	0
dl	0
loc	17
ccs	9
cts	9
cp	1
crap	4
rs	9.9332
c	0
b	0
f	0

<?php
/**
 * Plasma Driver MySQL component
 * Copyright 2018-2019 PlasmaPHP, All Rights Reserved
 *
 * Website: https://github.com/PlasmaPHP
 * License: https://github.com/PlasmaPHP/driver-mysql/blob/master/LICENSE
*/

namespace Plasma\Drivers\MySQL;

/**
 * The MySQL Protocol Parser.
 * @internal
 */
class ProtocolParser implements \Evenement\EventEmitterInterface {
    use \Evenement\EventEmitterTrait;
    
    /**
     * @var int
     */
    const STATE_INIT = 0;
    
    /**
     * @var int
     */
    const STATE_HANDSHAKE = 1;
    
    /**
     * @var int
     */
    const STATE_HANDSHAKE_ERROR = 2;
    
    /**
     * @var int
     */
    const STATE_AUTH = 5;
    
    /**
     * @var int
     */
    const STATE_AUTH_SENT = 6;
    
    /**
     * @var int
     */
    const STATE_AUTH_ERROR = 7;
    
    /**
     * @var int
     */
    const STATE_OK = 9;
    
    /**
     * @var int
     */
    const CLIENT_CAPABILITIES = (
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_FOUND_ROWS |
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_LONG_PASSWORD |
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_LONG_FLAG |
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_LOCAL_FILES |
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_INTERACTIVE |
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_TRANSACTIONS |
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_SECURE_CONNECTION |
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_PROTOCOL_41 |
        \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_DEPRECATE_EOF
    );
    
    /**
     * @var int
     */
    const CLIENT_MAX_PACKET_SIZE = 0xFFFFFF;
    
    /**
     * @var int
     */
    const CLIENT_CHARSET_NUMBER = 0x21;
    
    /**
     * @var \Plasma\Drivers\MySQL\Driver
     */
    protected $driver;
    
    /**
     * @var \React\Socket\ConnectionInterface
     */
    protected $connection;
    
    /**
     * @var int
     */
    protected $state = ProtocolParser::STATE_INIT;
    
    /**
     * @var \Plasma\BinaryBuffer
     */
    protected $buffer;
    
    /**
     * @var \Plasma\BinaryBuffer
     */
    protected $messageBuffer;
    
    /**
     * The sequence ID is incremented with each packet and may wrap around.
     * It starts at 0 and is reset to 0 when a new command begins in the Command Phase.
     * @var int
     * @see https://dev.mysql.com/doc/internals/en/sequence-id.html
     */
    protected $sequenceID = -1;
    
    /**
     * Whether we use compression.
     * @var bool
     */
    protected $compressionEnabled = false;
    
    /**
     * The compression ID is incremented with each packet and may wrap around.
     * The compression ID is independent to the sequence ID.
     * @var int
     * @see https://dev.mysql.com/doc/internals/en/compressed-packet-header.html
     */
    protected $compressionID = -1;
    
    /**
     * Small packets should not be compressed. This defines a minimum size for compression.
     * @var int
     * @see https://dev.mysql.com/doc/internals/en/uncompressed-payload.html
     */
    protected $compressionSizeThreshold = 50;
    
    /**
     * @var \Plasma\BinaryBuffer
     */
    protected $compressionBuffer;
    
    /**
     * @var \Plasma\Drivers\MySQL\Messages\HandshakeMessage|null
     */
    protected $handshakeMessage;
    
    /**
     * @var \Plasma\Drivers\MySQL\Messages\OkResponseMessage|null
     */
    protected $lastOkMessage;
    
    /**
     * @var \Plasma\CommandInterface|null
     */
    protected $currentCommand;
    
    /**
     * @var callable|null
     */
    protected $parseCallback;
    
    /**
     * Constructor.
     * @param \Plasma\Drivers\MySQL\Driver       $driver
     * @param \React\Socket\ConnectionInterface  $connection
     */
    function __construct(\Plasma\Drivers\MySQL\Driver $driver, \React\Socket\ConnectionInterface $connection) {
        $this->driver = $driver;
        $this->connection = $connection;
        
        $this->buffer = new \Plasma\BinaryBuffer();
        $this->messageBuffer = new \Plasma\BinaryBuffer();
        $this->compressionBuffer = new \Plasma\BinaryBuffer();
        
        $this->addEvents();
    }
    
    /**
     * Invoke a command to execute.
     * @param \Plasma\CommandInterface|null  $command
     * @return void
     */
    function invokeCommand(?\Plasma\CommandInterface $command): void {
        if($command === null) {
            return;
        }
        
        $this->currentCommand = $command;
        $this->processCommand();
    }
    
    /**
     * Executes a command, without handling any aftermath.
     * The `onComplete` callback will be immediately invoked, regardless of the `waitForCompletion` value.
     * @param \Plasma\CommandInterface  $command
     * @return void
     */
    function executeCommand(\Plasma\CommandInterface $command): void {
        $this->processCommand($command);
    }
    
    /**
     * Marks the command itself as finished, if currently running.
     * @param \Plasma\Drivers\MySQL\Commands\CommandInterface  $command
     * @return void
     */
    function markCommandAsFinished(\Plasma\CommandInterface $command): void {
        if($command === $this->currentCommand) {
            $this->currentCommand = null;
        }
        
        $command->onComplete();
    }
    
    /**
     * Get the parser state.
     * @return int
     */
    function getState(): int {
        return $this->state;
    }
    
    /**
     * Get the handshake message, or null.
     * @return \Plasma\Drivers\MySQL\Messages\HandshakeMessage|null
     */
    function getHandshakeMessage(): ?\Plasma\Drivers\MySQL\Messages\HandshakeMessage {
        return $this->handshakeMessage;
    }
    
    /**
     * Get the last ok response message, or null.
     * @return \Plasma\Drivers\MySQL\Messages\OkResponseMessage|null
     */
    function getLastOkMessage(): ?\Plasma\Drivers\MySQL\Messages\OkResponseMessage {
        return $this->lastOkMessage;
    }
    
    /**
     * Enables compression.
     * @return void
     */
    function enableCompression(): void {
        $this->compressionEnabled = true;
    }
    
    /**
     * Sends a packet to the server.
     * @param string  $packet
     * @return void
     */
    function sendPacket(string $packet): void {
        $initPacklen = \strlen($packet);
        
        do {
            $partial = \substr($packet, 0, static::CLIENT_MAX_PACKET_SIZE);
            $partlen = \strlen($partial);
            
            $packet = \substr($packet, static::CLIENT_MAX_PACKET_SIZE);
            $packlen = \strlen($packet);
            
            $length = \Plasma\BinaryBuffer::writeInt3($partlen);
            $sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
            
            $packet = $length.$sequence.$partial;
            
            if($this->compressionEnabled && $this->state === static::STATE_OK) {
                $packet = $this->compressPacket($packet);
            }
            
            $this->connection->write($packet);
        } while($packlen > static::CLIENT_MAX_PACKET_SIZE);
        
        // If the packet is exactly the max size, we have to send two packets
        if($initPacklen === static::CLIENT_MAX_PACKET_SIZE) {
            $length = \Plasma\BinaryBuffer::writeInt3(0);
            $sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
            $packet = $length.$sequence;
            
            if($this->compressionEnabled && $this->state === static::STATE_OK) {
                $packet = $this->compressPacket($packet);
            }
            
            $this->connection->write($packet);
        }
    }
    
    /**
     * Sets the parse callback.
     * @param callable $callback
     * @return void
     */
    function setParseCallback(callable $callback): void {
        $this->parseCallback = $callback;
    }
    
    /**
     * Processes a command.
     * @param \Plasma\CommandInterface|null  $command
     * @return void
     */
    protected function processCommand(?\Plasma\CommandInterface $command = null) {
        if($command === null && $this->currentCommand instanceof \Plasma\CommandInterface) {
            $command = $this->currentCommand;
            
            if($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) {
                $state = $command->setParserState();
                if($state !== -1) {
                    $this->state = $state;
                }
            }
        }
        
        if($command === null) {
            return;
        }
        
        if(!($command instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) || $command->resetSequence()) {
            $this->sequenceID = -1;
            $this->compressionID = -1;
        }
        
        $this->sendPacket($command->getEncodedMessage());
        
        if($command !== $this->currentCommand || !$command->waitForCompletion()) {
            $command->onComplete();
            
            if($command === $this->currentCommand) {
                $this->currentCommand = null;
            }
        }
    }
    
    /**
     * Processes the buffer.
     * @return void
     */
    protected function processBuffer() {
        if($this->buffer->getSize() < 4) {
            return;
        }
        
        $buffer = clone $this->buffer;
        
        $length = $buffer->readInt3();
        $this->sequenceID = $buffer->readInt1();
        
        if($length === static::CLIENT_MAX_PACKET_SIZE) {
            $this->buffer->read(($length + 4));
            $this->messageBuffer->append($buffer->read($length));
            return;
        } elseif($this->messageBuffer->getSize() > 0) {
            $this->messageBuffer->append($buffer->read($length));
            $buffer = $this->messageBuffer;
            $this->messageBuffer = new \Plasma\BinaryBuffer();
        }
        
        if($buffer->getSize() < $length) {
            return;
        }
        
        if($length > 0) {
            $this->buffer->read(($length + 4));
            $buffer->slice(0, $length);
        } else {
            $this->buffer->slice($buffer->getSize());
        }
        
        if($buffer->getSize() === 0) {
            return;
        }
        
        /** @var \Plasma\Drivers\MySQL\Messages\MessageInterface  $message */
        $message = null;
        
        if($this->state === static::STATE_INIT) {
            $message = new \Plasma\Drivers\MySQL\Messages\HandshakeMessage($this);
        } else {
            $firstChar = $buffer->read(1);
            
            $okRespID = \Plasma\Drivers\MySQL\Messages\OkResponseMessage::getID();
            $isOkMessage = (
                (
                    $firstChar === $okRespID &&
                    (!($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\QueryCommand)
                        || \strtoupper(\substr($this->currentCommand->getQuery(), 0, 6)) !== 'SELECT') // Fix for MySQL 5.7
                ) ||
                (
                    $firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() &&
                    ($this->handshakeMessage->capability & \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_DEPRECATE_EOF) !== 0
                )
            );
            
            switch(true) {
                case ($firstChar === \Plasma\Drivers\MySQL\Messages\ErrResponseMessage::getID()):
                    $message = new \Plasma\Drivers\MySQL\Messages\ErrResponseMessage($this);
                break;
                case ($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\StatementPrepareCommand && $firstChar === $okRespID):
                    $message = new \Plasma\Drivers\MySQL\Messages\PrepareStatementOkMessage($this);
                break;
                case $isOkMessage:
                    $message = new \Plasma\Drivers\MySQL\Messages\OkResponseMessage($this);
                    $this->lastOkMessage = $message;
                break;

                case ($this->state < static::STATE_OK && $firstChar === \Plasma\Drivers\MySQL\Messages\AuthMoreDataMessage::getID()):

                    $message = new \Plasma\Drivers\MySQL\Messages\AuthMoreDataMessage($this);
                break;

                case ($this->state < static::STATE_OK && $firstChar === \Plasma\Drivers\MySQL\Messages\AuthSwitchRequestMessage::getID()):

                    $message = new \Plasma\Drivers\MySQL\Messages\AuthSwitchRequestMessage($this);
                break;

                case ($this->state < static::STATE_OK && $firstChar === \Plasma\Drivers\MySQL\Messages\AuthMoreDataMessage::getID()):

                    $message = new \Plasma\Drivers\MySQL\Messages\AuthMoreDataMessage($this);
                break;
                case ($firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() && $length < 6):
                    $message = new \Plasma\Drivers\MySQL\Messages\EOFMessage($this);
                break;

                case ($firstChar === \Plasma\Drivers\MySQL\Messages\LocalInFileRequestMessage::getID()):

                    if($this->driver->getOptions()['localInFile.enable']) {

                        $message = new \Plasma\Drivers\MySQL\Messages\LocalInFileRequestMessage($this);
                    } else {

                        $this->emit('error', array((new \Plasma\Exception('MySQL server requested a local file, but the driver options is disabled'))));
                        
                        if($this->buffer->getSize() > 0) {

                            $this->driver->getLoop()->futureTick(function () {
                                $this->processBuffer();
                            });

                        }

                        
                        return;
                    }

                break;

                default:
                    $buffer->prepend($firstChar);
                    
                    if($this->parseCallback !== null) {
                        $parse = $this->parseCallback;
                        $this->parseCallback = null;
                        
                        $caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
                        $parse($caller);
                    } elseif($this->currentCommand !== null) {
                        $command = $this->currentCommand;
                        
                        $caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
                        $command->onNext($caller);
                        
                        if($command->hasFinished()) {
                            $this->currentCommand = null;
                            $command->onComplete();
                        }
                    }
                    
                    if($this->buffer->getSize() > 0) {
                        $this->driver->getLoop()->futureTick(function () {
                            $this->processBuffer();
                        });
                    }
                    
                    return;
                break;
            }
        }
        
        $state = $message->setParserState();
        if($state !== -1) {
            $this->state = $state;
        }
        
        if($message instanceof \Plasma\Drivers\MySQL\Messages\HandshakeMessage) {
            $this->handshakeMessage = $message;
        }
        
        $this->handleMessage($buffer, $message);
    }
    
    /**
     * Handles an incoming message.
     * @param \Plasma\BinaryBuffer                             $buffer
     * @param \Plasma\Drivers\MySQL\Messages\MessageInterface  $message
     * @return void
     */
    function handleMessage(\Plasma\BinaryBuffer $buffer, \Plasma\Drivers\MySQL\Messages\MessageInterface $message) {
        try {
            $buffer = $message->parseMessage($buffer);
            if(!$buffer) {
                return;
            }
            
            if($this->currentCommand !== null) {
                if(
                    ($message instanceof \Plasma\Drivers\MySQL\Messages\OkResponseMessage || $message instanceof \Plasma\Drivers\MySQL\Messages\EOFMessage)
                    && $this->currentCommand->hasFinished()
                ) {
                    $command = $this->currentCommand;
                    $this->currentCommand = null;
                    
                    $command->onComplete();
                } elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
                    $error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
                    
                    $command = $this->currentCommand;
                    $this->currentCommand = null;
                    
                    $command->onError($error);
                } else {
                    $command = $this->currentCommand;
                    $command->onNext($message);
                    
                    if($command->hasFinished()) {
                        if($this->currentCommand === $command) {
                            $this->currentCommand = null;
                        }
                        
                        $command->onComplete();
                    }
                }
            } elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
                $error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
                $this->emit('error', array($error));
            }
            
            $this->emit('message', array($message));
        } catch (\Plasma\Drivers\MySQL\Messages\ParseException $e) {
            $state = $e->getState();
            if($state !== null) {
                $this->state = $state;
            }
            
            $buffer = $e->getBuffer();
            if($buffer !== null) {
                $this->buffer->clear();
                $this->buffer->append($buffer);
            }
            
            if($this->currentCommand !== null) {
                $this->currentCommand->onError($e);
            }
            
            $this->emit('error', array($e));
            $this->connection->close();
        }
        
        if($this->buffer->getSize() > 0) {
            $this->driver->getLoop()->futureTick(function () {
                $this->processBuffer();
            });
        }
    }
    
    /**
     * Compresses a packet.
     * @param string  $packet
     * @return string
     */
    protected function compressPacket(string $packet): string {
        $length = \strlen($packet);
        $packetlen = \Plasma\BinaryBuffer::writeInt3($length);
        $id = \Plasma\BinaryBuffer::writeInt1((++$this->compressionID));
        
        if($length < $this->compressionSizeThreshold) {
            return $packetlen.$id.\Plasma\BinaryBuffer::writeInt3(0).$packet;
        }
        
        $compressed = \zlib_encode($packet, \ZLIB_ENCODING_DEFLATE);
        $compresslen = \Plasma\BinaryBuffer::writeInt3(\strlen($compressed));
        
        return $compresslen.$id.$compressed;
    }
    
    /**
     * Decompresses the buffer.
     * @return void
     */
    protected function decompressBuffer(): void {
        $buffer = new \Plasma\BinaryBuffer();
        
        // Copy packet header to new buffer
        for($i = 0; $i < 7; $i++) {
            $buffer->append($this->compressionBuffer[$i]);
        }
        
        $length = $buffer->readInt3();
        $this->compressionID = $buffer->readInt1();
        $uncompressedLength = $buffer->readInt3();
        
        if(($this->compressionBuffer->getSize() - 7) < $length) {
            return;
        }
        
        $this->compressionBuffer->read(7);
        $buffer = null;
        
        if($uncompressedLength === 0) {
            $this->buffer->append($this->compressionBuffer->read($length));
            return;
        }
        
        $rawPacket = $this->compressionBuffer->read($length);
        $packet = \zlib_decode($rawPacket, $uncompressedLength);
        
        if(\strlen($packet) !== $uncompressedLength) {
            $packet = "\xFF\x00\x00\x00     Invalid compressed packet";
            $this->connection->end($packet);
            
            return;
        }
        
        $this->buffer->append($packet);
        
        if($this->compressionBuffer->getSize() > 7) {
            $this->decompressBuffer();
        }
    }
    
    /**
     * Adds the events to the connection.
     * @return void
     */
    protected function addEvents() {
        $this->connection->on('data', function ($chunk) {
            if($this->compressionEnabled && $this->state === static::STATE_OK) {
                $this->compressionBuffer->append($chunk);
                
                if($this->compressionBuffer->getSize() > 7) {
                    $this->decompressBuffer();
                }
            } else {
                $this->buffer->append($chunk);
            }
            
            $this->processBuffer();
        });
        
        $this->connection->on('close', function () {
            $this->handleClose();
        });
    }
    
    /**
     * Connection close handler.
     * @return void
     */
    protected function handleClose() {
        if($this->state === static::STATE_AUTH || $this->state === static::STATE_AUTH_SENT) {
            $this->state = static::STATE_AUTH_ERROR;
        }
        
        $this->buffer->clear();
        $this->messageBuffer->clear();
        $this->compressionBuffer->clear();
    }
}


1		<?php
2		/**
3		* Plasma Driver MySQL component
4		* Copyright 2018-2019 PlasmaPHP, All Rights Reserved
5		*
6		* Website: https://github.com/PlasmaPHP
7		* License: https://github.com/PlasmaPHP/driver-mysql/blob/master/LICENSE
8		*/
9
10		namespace Plasma\Drivers\MySQL;
11
12		/**
13		* The MySQL Protocol Parser.
14		* @internal
15		*/
16		class ProtocolParser implements \Evenement\EventEmitterInterface {
17		use \Evenement\EventEmitterTrait;
18
19		/**
20		* @var int
21		*/
22		const STATE_INIT = 0;
23
24		/**
25		* @var int
26		*/
27		const STATE_HANDSHAKE = 1;
28
29		/**
30		* @var int
31		*/
32		const STATE_HANDSHAKE_ERROR = 2;
33
34		/**
35		* @var int
36		*/
37		const STATE_AUTH = 5;
38
39		/**
40		* @var int
41		*/
42		const STATE_AUTH_SENT = 6;
43
44		/**
45		* @var int
46		*/
47		const STATE_AUTH_ERROR = 7;
48
49		/**
50		* @var int
51		*/
52		const STATE_OK = 9;
53
54		/**
55		* @var int
56		*/
57		const CLIENT_CAPABILITIES = (
58		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_FOUND_ROWS \|
59		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_LONG_PASSWORD \|
60		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_LONG_FLAG \|
61		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_LOCAL_FILES \|
62		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_INTERACTIVE \|
63		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_TRANSACTIONS \|
64		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_SECURE_CONNECTION \|
65		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_PROTOCOL_41 \|
66		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_DEPRECATE_EOF
67		);
68
69		/**
70		* @var int
71		*/
72		const CLIENT_MAX_PACKET_SIZE = 0xFFFFFF;
73
74		/**
75		* @var int
76		*/
77		const CLIENT_CHARSET_NUMBER = 0x21;
78
79		/**
80		* @var \Plasma\Drivers\MySQL\Driver
81		*/
82		protected $driver;
83
84		/**
85		* @var \React\Socket\ConnectionInterface
86		*/
87		protected $connection;
88
89		/**
90		* @var int
91		*/
92		protected $state = ProtocolParser::STATE_INIT;
93
94		/**
95		* @var \Plasma\BinaryBuffer
96		*/
97		protected $buffer;
98
99		/**
100		* @var \Plasma\BinaryBuffer
101		*/
102		protected $messageBuffer;
103
104		/**
105		* The sequence ID is incremented with each packet and may wrap around.
106		* It starts at 0 and is reset to 0 when a new command begins in the Command Phase.
107		* @var int
108		* @see https://dev.mysql.com/doc/internals/en/sequence-id.html
109		*/
110		protected $sequenceID = -1;
111
112		/**
113		* Whether we use compression.
114		* @var bool
115		*/
116		protected $compressionEnabled = false;
117
118		/**
119		* The compression ID is incremented with each packet and may wrap around.
120		* The compression ID is independent to the sequence ID.
121		* @var int
122		* @see https://dev.mysql.com/doc/internals/en/compressed-packet-header.html
123		*/
124		protected $compressionID = -1;
125
126		/**
127		* Small packets should not be compressed. This defines a minimum size for compression.
128		* @var int
129		* @see https://dev.mysql.com/doc/internals/en/uncompressed-payload.html
130		*/
131		protected $compressionSizeThreshold = 50;
132
133		/**
134		* @var \Plasma\BinaryBuffer
135		*/
136		protected $compressionBuffer;
137
138		/**
139		* @var \Plasma\Drivers\MySQL\Messages\HandshakeMessage\|null
140		*/
141		protected $handshakeMessage;
142
143		/**
144		* @var \Plasma\Drivers\MySQL\Messages\OkResponseMessage\|null
145		*/
146		protected $lastOkMessage;
147
148		/**
149		* @var \Plasma\CommandInterface\|null
150		*/
151		protected $currentCommand;
152
153		/**
154		* @var callable\|null
155		*/
156		protected $parseCallback;
157
158		/**
159		* Constructor.
160		* @param \Plasma\Drivers\MySQL\Driver $driver
161		* @param \React\Socket\ConnectionInterface $connection
162		*/
163	63	function __construct(\Plasma\Drivers\MySQL\Driver $driver, \React\Socket\ConnectionInterface $connection) {
164	63	$this->driver = $driver;
165	63	$this->connection = $connection;
166
167	63	$this->buffer = new \Plasma\BinaryBuffer();
168	63	$this->messageBuffer = new \Plasma\BinaryBuffer();
169	63	$this->compressionBuffer = new \Plasma\BinaryBuffer();
170
171	63	$this->addEvents();
172	63	}
173
174		/**
175		* Invoke a command to execute.
176		* @param \Plasma\CommandInterface\|null $command
177		* @return void
178		*/
179	62	function invokeCommand(?\Plasma\CommandInterface $command): void {
180	62	if($command === null) {
181	59	return;
182		}
183
184	62	$this->currentCommand = $command;
185	62	$this->processCommand();
186	62	}
187
188		/**
189		* Executes a command, without handling any aftermath.
190		* The `onComplete` callback will be immediately invoked, regardless of the `waitForCompletion` value.
191		* @param \Plasma\CommandInterface $command
192		* @return void
193		*/
194		function executeCommand(\Plasma\CommandInterface $command): void {
195		$this->processCommand($command);
196		}
197
198		/**
199		* Marks the command itself as finished, if currently running.
200		* @param \Plasma\Drivers\MySQL\Commands\CommandInterface $command
201		* @return void
202		*/
203	59	function markCommandAsFinished(\Plasma\CommandInterface $command): void {
204	59	if($command === $this->currentCommand) {
205	59	$this->currentCommand = null;
206		}
207
208	59	$command->onComplete();
209	59	}
210
211		/**
212		* Get the parser state.
213		* @return int
214		*/
215	1	function getState(): int {
216	1	return $this->state;
217		}
218
219		/**
220		* Get the handshake message, or null.
221		* @return \Plasma\Drivers\MySQL\Messages\HandshakeMessage\|null
222		*/
223	62	function getHandshakeMessage(): ?\Plasma\Drivers\MySQL\Messages\HandshakeMessage {
224	62	return $this->handshakeMessage;
225		}
226
227		/**
228		* Get the last ok response message, or null.
229		* @return \Plasma\Drivers\MySQL\Messages\OkResponseMessage\|null
230		*/
231	4	function getLastOkMessage(): ?\Plasma\Drivers\MySQL\Messages\OkResponseMessage {
232	4	return $this->lastOkMessage;
233		}
234
235		/**
236		* Enables compression.
237		* @return void
238		*/
239	2	function enableCompression(): void {
240	2	$this->compressionEnabled = true;
241	2	}
242
243		/**
244		* Sends a packet to the server.
245		* @param string $packet
246		* @return void
247		*/
248	63	function sendPacket(string $packet): void {
249	63	$initPacklen = \strlen($packet);
250
251		do {
252	63	$partial = \substr($packet, 0, static::CLIENT_MAX_PACKET_SIZE);
253	63	$partlen = \strlen($partial);
254
255	63	$packet = \substr($packet, static::CLIENT_MAX_PACKET_SIZE);
256	63	$packlen = \strlen($packet);
257
258	63	$length = \Plasma\BinaryBuffer::writeInt3($partlen);
259	63	$sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
260
261	63	$packet = $length.$sequence.$partial;
262
263	63	if($this->compressionEnabled && $this->state === static::STATE_OK) {
264	2	$packet = $this->compressPacket($packet);
265		}
266
267	63	$this->connection->write($packet);
268	63	} while($packlen > static::CLIENT_MAX_PACKET_SIZE);
269
270		// If the packet is exactly the max size, we have to send two packets
271	63	if($initPacklen === static::CLIENT_MAX_PACKET_SIZE) {
272	1	$length = \Plasma\BinaryBuffer::writeInt3(0);
273	1	$sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
274	1	$packet = $length.$sequence;
275
276	1	if($this->compressionEnabled && $this->state === static::STATE_OK) {
277		$packet = $this->compressPacket($packet);
278		}
279
280	1	$this->connection->write($packet);
281		}
282	63	}
283
284		/**
285		* Sets the parse callback.
286		* @param callable $callback
287		* @return void
288		*/
289		function setParseCallback(callable $callback): void {
290		$this->parseCallback = $callback;
291		}
292
293		/**
294		* Processes a command.
295		* @param \Plasma\CommandInterface\|null $command
296		* @return void
297		*/
298	62	protected function processCommand(?\Plasma\CommandInterface $command = null) {
299	62	if($command === null && $this->currentCommand instanceof \Plasma\CommandInterface) {
300	62	$command = $this->currentCommand;
301
302	62	if($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) {
303	62	$state = $command->setParserState();
304	62	if($state !== -1) {
305	62	$this->state = $state;
306		}
307		}
308		}
309
310	62	if($command === null) {
311		return;
312		}
313
314	62	if(!($command instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) \|\| $command->resetSequence()) {
315	59	$this->sequenceID = -1;
316	59	$this->compressionID = -1;
317		}
318
319	62	$this->sendPacket($command->getEncodedMessage());
320
321	62	if($command !== $this->currentCommand \|\| !$command->waitForCompletion()) {
322	35	$command->onComplete();
323
324	35	if($command === $this->currentCommand) {
325	35	$this->currentCommand = null;
326		}
327		}
328	62	}
329
330		/**
331		* Processes the buffer.
332		* @return void
333		*/
334	62	protected function processBuffer() {
335	62	if($this->buffer->getSize() < 4) {
336		return;
337		}
338
339	62	$buffer = clone $this->buffer;
340
341	62	$length = $buffer->readInt3();
342	62	$this->sequenceID = $buffer->readInt1();
343
344	62	if($length === static::CLIENT_MAX_PACKET_SIZE) {
345		$this->buffer->read(($length + 4));
346		$this->messageBuffer->append($buffer->read($length));
347		return;
348	62	} elseif($this->messageBuffer->getSize() > 0) {
349		$this->messageBuffer->append($buffer->read($length));
350		$buffer = $this->messageBuffer;
351		$this->messageBuffer = new \Plasma\BinaryBuffer();
352		}
353
354	62	if($buffer->getSize() < $length) {
355		return;
356		}
357
358	62	if($length > 0) {
359	62	$this->buffer->read(($length + 4));
360	62	$buffer->slice(0, $length);
361		} else {
362		$this->buffer->slice($buffer->getSize());
363		}
364
365	62	if($buffer->getSize() === 0) {
366		return;
367		}
368
369		/** @var \Plasma\Drivers\MySQL\Messages\MessageInterface $message */
370	62	$message = null;
371
372	62	if($this->state === static::STATE_INIT) {
373	62	$message = new \Plasma\Drivers\MySQL\Messages\HandshakeMessage($this);
374		} else {
375	62	$firstChar = $buffer->read(1);
376
377	62	$okRespID = \Plasma\Drivers\MySQL\Messages\OkResponseMessage::getID();
378		$isOkMessage = (
379		(
380	62	$firstChar === $okRespID &&
381	61	(!($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\QueryCommand)
382	61	\|\| \strtoupper(\substr($this->currentCommand->getQuery(), 0, 6)) !== 'SELECT') // Fix for MySQL 5.7
383		) \|\|
384		(
385	48	$firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() &&
386	62	($this->handshakeMessage->capability & \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_DEPRECATE_EOF) !== 0
387		)
388		);
389
390		switch(true) {
391	62	case ($firstChar === \Plasma\Drivers\MySQL\Messages\ErrResponseMessage::getID()):
392	1	$message = new \Plasma\Drivers\MySQL\Messages\ErrResponseMessage($this);
393	1	break;
394	61	case ($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\StatementPrepareCommand && $firstChar === $okRespID):
395	37	$message = new \Plasma\Drivers\MySQL\Messages\PrepareStatementOkMessage($this);
396	37	break;
397	61	case $isOkMessage:
398	61	$message = new \Plasma\Drivers\MySQL\Messages\OkResponseMessage($this);
399	61	$this->lastOkMessage = $message;
400	61	break;
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Case breaking statement indented incorrectly; expected 20 spaces, found 16 Loading history...
401	47	case ($this->state < static::STATE_OK && $firstChar === \Plasma\Drivers\MySQL\Messages\AuthMoreDataMessage::getID()):
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 12 spaces, found 16 Loading history...
402		$message = new \Plasma\Drivers\MySQL\Messages\AuthMoreDataMessage($this);
403		break;
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Case breaking statement indented incorrectly; expected 20 spaces, found 16 Loading history...
404	47	case ($this->state < static::STATE_OK && $firstChar === \Plasma\Drivers\MySQL\Messages\AuthSwitchRequestMessage::getID()):
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 12 spaces, found 16 Loading history...
405		$message = new \Plasma\Drivers\MySQL\Messages\AuthSwitchRequestMessage($this);
406		break;
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Case breaking statement indented incorrectly; expected 20 spaces, found 16 Loading history...
407	47	case ($this->state < static::STATE_OK && $firstChar === \Plasma\Drivers\MySQL\Messages\AuthMoreDataMessage::getID()):
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 12 spaces, found 16 Loading history...
408		$message = new \Plasma\Drivers\MySQL\Messages\AuthMoreDataMessage($this);
409		break;
410	47	case ($firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() && $length < 6):
411		$message = new \Plasma\Drivers\MySQL\Messages\EOFMessage($this);
412		break;
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Case breaking statement indented incorrectly; expected 20 spaces, found 16 Loading history...
413	47	case ($firstChar === \Plasma\Drivers\MySQL\Messages\LocalInFileRequestMessage::getID()):
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 12 spaces, found 16 Loading history...
414		if($this->driver->getOptions()['localInFile.enable']) {
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 16 spaces, found 20 Loading history...
415		$message = new \Plasma\Drivers\MySQL\Messages\LocalInFileRequestMessage($this);
416		} else {
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 16 spaces, found 20 Loading history...
417		$this->emit('error', array((new \Plasma\Exception('MySQL server requested a local file, but the driver options is disabled'))));
418
419		if($this->buffer->getSize() > 0) {
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 20 spaces, found 24 Loading history...
420		$this->driver->getLoop()->futureTick(function () {
421		$this->processBuffer();
422		});
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 24 spaces, found 28 Loading history...
423		}
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 20 spaces, found 24 Loading history...
424
425		return;
426		}
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Line indented incorrectly; expected 16 spaces, found 20 Loading history...
427		break;
		1 ignored issue – show Coding Style introduced 2019-03-18 19:39 UTC by Report Bug Copy Issue Report Case breaking statement indented incorrectly; expected 20 spaces, found 16 Loading history...
428		default:
429	47	$buffer->prepend($firstChar);
430
431	47	if($this->parseCallback !== null) {
432		$parse = $this->parseCallback;
433		$this->parseCallback = null;
434
435		$caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
436		$parse($caller);
437	47	} elseif($this->currentCommand !== null) {
438	47	$command = $this->currentCommand;
439
440	47	$caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
441	47	$command->onNext($caller);
442
443	47	if($command->hasFinished()) {
444	37	$this->currentCommand = null;
445	37	$command->onComplete();
446		}
447		}
448
449	47	if($this->buffer->getSize() > 0) {
450		$this->driver->getLoop()->futureTick(function () {
451	47	$this->processBuffer();
452	47	});
453		}
454
455	47	return;
456		break;
457		}
458		}
459
460	62	$state = $message->setParserState();
461	62	if($state !== -1) {
462	62	$this->state = $state;
463		}
464
465	62	if($message instanceof \Plasma\Drivers\MySQL\Messages\HandshakeMessage) {
466	62	$this->handshakeMessage = $message;
467		}
468
469	62	$this->handleMessage($buffer, $message);
470	62	}
471
472		/**
473		* Handles an incoming message.
474		* @param \Plasma\BinaryBuffer $buffer
475		* @param \Plasma\Drivers\MySQL\Messages\MessageInterface $message
476		* @return void
477		*/
478	62	function handleMessage(\Plasma\BinaryBuffer $buffer, \Plasma\Drivers\MySQL\Messages\MessageInterface $message) {
479		try {
480	62	$buffer = $message->parseMessage($buffer);
481	62	if(!$buffer) {
482		return;
483		}
484
485	62	if($this->currentCommand !== null) {
486		if(
487	62	($message instanceof \Plasma\Drivers\MySQL\Messages\OkResponseMessage \|\| $message instanceof \Plasma\Drivers\MySQL\Messages\EOFMessage)
488	62	&& $this->currentCommand->hasFinished()
489		) {
490	61	$command = $this->currentCommand;
491	61	$this->currentCommand = null;
492
493	61	$command->onComplete();
494	60	} elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
495	1	$error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
496
497	1	$command = $this->currentCommand;
498	1	$this->currentCommand = null;
499
500	1	$command->onError($error);
501		} else {
502	59	$command = $this->currentCommand;
503	59	$command->onNext($message);
504
505	59	if($command->hasFinished()) {
506	59	if($this->currentCommand === $command) {
507		$this->currentCommand = null;
508		}
509
510	62	$command->onComplete();
511		}
512		}
513	62	} elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
514		$error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
515		$this->emit('error', array($error));
516		}
517
518	62	$this->emit('message', array($message));
519		} catch (\Plasma\Drivers\MySQL\Messages\ParseException $e) {
520		$state = $e->getState();
521		if($state !== null) {
522		$this->state = $state;
523		}
524
525		$buffer = $e->getBuffer();
526		if($buffer !== null) {
527		$this->buffer->clear();
528		$this->buffer->append($buffer);
529		}
530
531		if($this->currentCommand !== null) {
532		$this->currentCommand->onError($e);
533		}
534
535		$this->emit('error', array($e));
536		$this->connection->close();
537		}
538
539	62	if($this->buffer->getSize() > 0) {
540		$this->driver->getLoop()->futureTick(function () {
541	37	$this->processBuffer();
542	37	});
543		}
544	62	}
545
546		/**
547		* Compresses a packet.
548		* @param string $packet
549		* @return string
550		*/
551	2	protected function compressPacket(string $packet): string {
552	2	$length = \strlen($packet);
553	2	$packetlen = \Plasma\BinaryBuffer::writeInt3($length);
554	2	$id = \Plasma\BinaryBuffer::writeInt1((++$this->compressionID));
555
556	2	if($length < $this->compressionSizeThreshold) {
557	2	return $packetlen.$id.\Plasma\BinaryBuffer::writeInt3(0).$packet;
558		}
559
560		$compressed = \zlib_encode($packet, \ZLIB_ENCODING_DEFLATE);
561		$compresslen = \Plasma\BinaryBuffer::writeInt3(\strlen($compressed));
562
563		return $compresslen.$id.$compressed;
564		}
565
566		/**
567		* Decompresses the buffer.
568		* @return void
569		*/
570	2	protected function decompressBuffer(): void {
571	2	$buffer = new \Plasma\BinaryBuffer();
572
573		// Copy packet header to new buffer
574	2	for($i = 0; $i < 7; $i++) {
575	2	$buffer->append($this->compressionBuffer[$i]);
576		}
577
578	2	$length = $buffer->readInt3();
579	2	$this->compressionID = $buffer->readInt1();
580	2	$uncompressedLength = $buffer->readInt3();
581
582	2	if(($this->compressionBuffer->getSize() - 7) < $length) {
583		return;
584		}
585
586	2	$this->compressionBuffer->read(7);
587	2	$buffer = null;
588
589	2	if($uncompressedLength === 0) {
590	2	$this->buffer->append($this->compressionBuffer->read($length));
591	2	return;
592		}
593
594	1	$rawPacket = $this->compressionBuffer->read($length);
595	1	$packet = \zlib_decode($rawPacket, $uncompressedLength);
596
597	1	if(\strlen($packet) !== $uncompressedLength) {
598		$packet = "\xFF\x00\x00\x00 Invalid compressed packet";
599		$this->connection->end($packet);
600
601		return;
602		}
603
604	1	$this->buffer->append($packet);
605
606	1	if($this->compressionBuffer->getSize() > 7) {
607		$this->decompressBuffer();
608		}
609	1	}
610
611		/**
612		* Adds the events to the connection.
613		* @return void
614		*/
615	63	protected function addEvents() {
616		$this->connection->on('data', function ($chunk) {
617	62	if($this->compressionEnabled && $this->state === static::STATE_OK) {
618	2	$this->compressionBuffer->append($chunk);
619
620	2	if($this->compressionBuffer->getSize() > 7) {
621	2	$this->decompressBuffer();
622		}
623		} else {
624	62	$this->buffer->append($chunk);
625		}
626
627	62	$this->processBuffer();
628	63	});
629
630		$this->connection->on('close', function () {
631	3	$this->handleClose();
632	63	});
633	63	}
634
635		/**
636		* Connection close handler.
637		* @return void
638		*/
639	3	protected function handleClose() {
640	3	if($this->state === static::STATE_AUTH \|\| $this->state === static::STATE_AUTH_SENT) {
641		$this->state = static::STATE_AUTH_ERROR;
642		}
643
644	3	$this->buffer->clear();
645	3	$this->messageBuffer->clear();
646	3	$this->compressionBuffer->clear();
647	3	}
648		}
649

PlasmaPHP / driver-mysql

GitHub Access Token became invalid

Push — master ( 51f19d...b0fd42 )

ProtocolParser::addEvents() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like