ProtocolParser::getHandshakeMessage() - Code Metrics - Inspection of "Fix bug and comment out asserts" - PlasmaPHP/driver-mysql - Measure and Improve Code Quality continuously with Scrutinizer

Failed Conditions

Push — master ( 396319...5c4ceb )

by Charlotte

created 2018-11-18 15:08 UTC

ProtocolParser::getHandshakeMessage() A

↳ Parent: ProtocolParser

Complexity

Conditions	1
Paths	1

Size

Total Lines	2
Code Lines	1

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

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

<?php
/**
 * Plasma Driver MySQL component
 * Copyright 2018 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_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 = 0x1000000;
    
    /**
     * @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 = 0;
    
    /**
     * @var \Plasma\Drivers\MySQL\Messages\HandshakeMessage|null
     */
    protected $handshakeMessage;
    
    /**
     * @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->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;
    }
    
    /**
     * Sends a packet to the server.
     * @param string  $packet
     * @return void
     */
    function sendPacket(string $packet): void {
        $maxSize = static::CLIENT_MAX_PACKET_SIZE - 4;
        
        do {
            $partial = \substr($packet, 0, $maxSize);
            $packet = \substr($packet, $maxSize);
            
            $length = \Plasma\BinaryBuffer::writeInt3(\strlen($partial));
            $sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
            
            $this->connection->write($length.$sequence.$partial);
        } while(\strlen($packet) > $maxSize);
    }
    
    /**
     * 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;
        }
        
        // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Processing command '.get_class($command)) || true));
        
        $this->sendPacket($command->getEncodedMessage());
        
        if($command !== $this->currentCommand || !$command->waitForCompletion()) {
            // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Mark command as completed') || true));
            $command->onComplete();
            
            if($command === $this->currentCommand) {
                $this->currentCommand = null;
            }
        }
    }
    
    /**
     * Processes the buffer.
     * @return void
     */
    protected function processBuffer() {
        // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('ProtocolParser::processBuffer called') || true));
        
        if($this->buffer->getSize() < 4) {
            // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Not enough data received for packet header ('.$this->buffer->getSize().')') || true));
            return;
        }
        
        $buffer = clone $this->buffer;
        
        $length = $buffer->readInt3();
        $this->sequenceID = $buffer->readInt1();
        
        // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('First 10 bytes: '.implode(', ', unpack('C*', \substr($this->buffer->getContents(), 0, 10)))) || true));
        // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Read packet header length ('.$length.') and sequence ('.$this->sequenceID.')') || true));
        
        if($length === 0xFFFFFF) {
            $this->buffer->read(($length + 4));
            $this->messageBuffer->append($buffer->read($length));
            
            // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned, 16mb packet received, waiting for the last one to arrive') || true));
            return;
        } elseif($this->messageBuffer->getSize() > 0) {
            $this->messageBuffer->append($buffer->read($length));
            $buffer = $this->messageBuffer;
            $this->messageBuffer = new \Plasma\BinaryBuffer();
        }
        
        if($buffer->getSize() < $length) {
            // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned, insufficent length: '.$buffer->getSize().', '.$length.' required') || true));
            return;
        }
        
        if($length > 0) {
            $this->buffer->read(($length + 4));
            $buffer->slice(0, $length);
        } else {
            $this->buffer->slice($buffer->getSize());
        }
        
        if($buffer->getSize() === 0) {
            // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Buffer length is 0') || true));
            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);
            // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Message char "'.$firstChar.'" (0x'.\dechex(\ord($firstChar)).') - buffer length: '.$buffer->getSize()) || true));
            
            $okRespID = \Plasma\Drivers\MySQL\Messages\OkResponseMessage::getID();
            $isOkMessage = (
                (
                    $firstChar === $okRespID &&
                    !($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\StatementExecuteCommand)
                ) ||
                (
                    $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);
                break;
                case ($firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() && $length < 6):
                    $message = new \Plasma\Drivers\MySQL\Messages\EOFMessage($this);
                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) {
                        $caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
                        $this->currentCommand->onNext($caller);
                    }
                    
                    // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Left over buffer: '.$buffer->getSize()) || true));
                    
                    if($this->buffer->getSize() > 0) {
                        // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Scheduling future read with '.$this->buffer->getSize().' bytes') || true));
                        
                        $this->driver->getLoop()->futureTick(function () {
                            $this->processBuffer();
                        });
                    }
                    
                    return;
                break;
            }
        }
        
        // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Message '.\get_class($message)) || true));
        
        $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) {
                // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned handle (unsufficent buffer length)') || true));
                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) {
                // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Error Response Message with message: '.$message->errorMessage) || true));
                
                $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) {
            // for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Scheduling future read (msg) with '.$this->buffer->getSize().' bytes') || true));
            
            $this->driver->getLoop()->futureTick(function () {
                $this->processBuffer();
            });
        }
    }
    
    /**
     * Adds the events to the connection.
     * @return void
     */
    protected function addEvents() {
        $this->connection->on('data', function ($chunk) {
            $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;
        }
    }
}


1		<?php
2		/**
3		* Plasma Driver MySQL component
4		* Copyright 2018 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_LONG_PASSWORD \|
59		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_LONG_FLAG \|
60		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_LOCAL_FILES \|
61		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_INTERACTIVE \|
62		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_TRANSACTIONS \|
63		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_SECURE_CONNECTION \|
64		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_PROTOCOL_41 \|
65		\Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_DEPRECATE_EOF
66		);
67
68		/**
69		* @var int
70		*/
71		const CLIENT_MAX_PACKET_SIZE = 0x1000000;
72
73		/**
74		* @var int
75		*/
76		const CLIENT_CHARSET_NUMBER = 0x21;
77
78		/**
79		* @var \Plasma\Drivers\MySQL\Driver
80		*/
81		protected $driver;
82
83		/**
84		* @var \React\Socket\ConnectionInterface
85		*/
86		protected $connection;
87
88		/**
89		* @var int
90		*/
91		protected $state = ProtocolParser::STATE_INIT;
92
93		/**
94		* @var \Plasma\BinaryBuffer
95		*/
96		protected $buffer;
97
98		/**
99		* @var \Plasma\BinaryBuffer
100		*/
101		protected $messageBuffer;
102
103		/**
104		* The sequence ID is incremented with each packet and may wrap around.
105		* It starts at 0 and is reset to 0 when a new command begins in the Command Phase.
106		* @var int
107		* @see https://dev.mysql.com/doc/internals/en/sequence-id.html
108		*/
109		protected $sequenceID = 0;
110
111		/**
112		* @var \Plasma\Drivers\MySQL\Messages\HandshakeMessage\|null
113		*/
114		protected $handshakeMessage;
115
116		/**
117		* @var \Plasma\CommandInterface\|null
118		*/
119		protected $currentCommand;
120
121		/**
122		* @var callable\|null
123		*/
124		protected $parseCallback;
125
126		/**
127		* Constructor.
128		* @param \Plasma\Drivers\MySQL\Driver $driver
129		* @param \React\Socket\ConnectionInterface $connection
130		*/
131	12	function __construct(\Plasma\Drivers\MySQL\Driver $driver, \React\Socket\ConnectionInterface $connection) {
132	12	$this->driver = $driver;
133	12	$this->connection = $connection;
134
135	12	$this->buffer = new \Plasma\BinaryBuffer();
136	12	$this->messageBuffer = new \Plasma\BinaryBuffer();
137
138	12	$this->addEvents();
139	12	}
140
141		/**
142		* Invoke a command to execute.
143		* @param \Plasma\CommandInterface\|null $command
144		* @return void
145		*/
146	12	function invokeCommand(?\Plasma\CommandInterface $command): void {
147	12	if($command === null) {
148	2	return;
149		}
150
151	12	$this->currentCommand = $command;
152	12	$this->processCommand();
153	12	}
154
155		/**
156		* Executes a command, without handling any aftermath.
157		* The `onComplete` callback will be immediately invoked, regardless of the `waitForCompletion` value.
158		* @param \Plasma\CommandInterface $command
159		* @return void
160		*/
161		function executeCommand(\Plasma\CommandInterface $command): void {
162		$this->processCommand($command);
163		}
164
165		/**
166		* Marks the command itself as finished, if currently running.
167		* @param \Plasma\Drivers\MySQL\Commands\CommandInterface $command
168		* @return void
169		*/
170	2	function markCommandAsFinished(\Plasma\CommandInterface $command): void {
171	2	if($command === $this->currentCommand) {
172	2	$this->currentCommand = null;
173		}
174
175	2	$command->onComplete();
176	2	}
177
178		/**
179		* Get the parser state.
180		* @return int
181		*/
182	2	function getState(): int {
183	2	return $this->state;
184		}
185
186		/**
187		* Get the handshake message, or null.
188		* @return \Plasma\Drivers\MySQL\Messages\HandshakeMessage\|null
189		*/
190	12	function getHandshakeMessage(): ?\Plasma\Drivers\MySQL\Messages\HandshakeMessage {
191	12	return $this->handshakeMessage;
192		}
193
194		/**
195		* Sends a packet to the server.
196		* @param string $packet
197		* @return void
198		*/
199	12	function sendPacket(string $packet): void {
200	12	$maxSize = static::CLIENT_MAX_PACKET_SIZE - 4;
201
202		do {
203	12	$partial = \substr($packet, 0, $maxSize);
204	12	$packet = \substr($packet, $maxSize);
205
206	12	$length = \Plasma\BinaryBuffer::writeInt3(\strlen($partial));
207	12	$sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
208
209	12	$this->connection->write($length.$sequence.$partial);
210	12	} while(\strlen($packet) > $maxSize);
211	12	}
212
213		/**
214		* Sets the parse callback.
215		* @param callable $callback
216		* @return void
217		*/
218		function setParseCallback(callable $callback): void {
219		$this->parseCallback = $callback;
220		}
221
222		/**
223		* Processes a command.
224		* @param \Plasma\CommandInterface\|null $command
225		* @return void
226		*/
227	12	protected function processCommand(?\Plasma\CommandInterface $command = null) {
228	12	if($command === null && $this->currentCommand instanceof \Plasma\CommandInterface) {
229	12	$command = $this->currentCommand;
230
231	12	if($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) {
232	12	$state = $command->setParserState();
233	12	if($state !== -1) {
234	12	$this->state = $state;
235		}
236		}
237		}
238
239	12	if($command === null) {
240		return;
241		}
242
243	12	if(!($command instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) \|\| $command->resetSequence()) {
244	8	$this->sequenceID = -1;
245		}
246
247		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Processing command '.get_class($command)) \|\| true));
248
249	12	$this->sendPacket($command->getEncodedMessage());
250
251	12	if($command !== $this->currentCommand \|\| !$command->waitForCompletion()) {
252		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Mark command as completed') \|\| true));
253		$command->onComplete();
254
255		if($command === $this->currentCommand) {
256		$this->currentCommand = null;
257		}
258		}
259	12	}
260
261		/**
262		* Processes the buffer.
263		* @return void
264		*/
265	12	protected function processBuffer() {
266		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('ProtocolParser::processBuffer called') \|\| true));
267
268	12	if($this->buffer->getSize() < 4) {
269		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Not enough data received for packet header ('.$this->buffer->getSize().')') \|\| true));
270		return;
271		}
272
273	12	$buffer = clone $this->buffer;
274
275	12	$length = $buffer->readInt3();
276	12	$this->sequenceID = $buffer->readInt1();
277
278		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('First 10 bytes: '.implode(', ', unpack('C*', \substr($this->buffer->getContents(), 0, 10)))) \|\| true));
279		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Read packet header length ('.$length.') and sequence ('.$this->sequenceID.')') \|\| true));
280
281	12	if($length === 0xFFFFFF) {
282		$this->buffer->read(($length + 4));
283		$this->messageBuffer->append($buffer->read($length));
284
285		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned, 16mb packet received, waiting for the last one to arrive') \|\| true));
286		return;
287	12	} elseif($this->messageBuffer->getSize() > 0) {
288		$this->messageBuffer->append($buffer->read($length));
289		$buffer = $this->messageBuffer;
290		$this->messageBuffer = new \Plasma\BinaryBuffer();
291		}
292
293	12	if($buffer->getSize() < $length) {
294		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned, insufficent length: '.$buffer->getSize().', '.$length.' required') \|\| true));
295		return;
296		}
297
298	12	if($length > 0) {
299	12	$this->buffer->read(($length + 4));
300	12	$buffer->slice(0, $length);
301		} else {
302		$this->buffer->slice($buffer->getSize());
303		}
304
305	12	if($buffer->getSize() === 0) {
306		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Buffer length is 0') \|\| true));
307		return;
308		}
309
310		/** @var \Plasma\Drivers\MySQL\Messages\MessageInterface $message */
311	12	$message = null;
312
313	12	if($this->state === static::STATE_INIT) {
314	12	$message = new \Plasma\Drivers\MySQL\Messages\HandshakeMessage($this);
315		} else {
316	12	$firstChar = $buffer->read(1);
317		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Message char "'.$firstChar.'" (0x'.\dechex(\ord($firstChar)).') - buffer length: '.$buffer->getSize()) \|\| true));
318
319	12	$okRespID = \Plasma\Drivers\MySQL\Messages\OkResponseMessage::getID();
320		$isOkMessage = (
321		(
322	12	$firstChar === $okRespID &&
323	10	!($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\StatementExecuteCommand)
324		) \|\|
325		(
326	6	$firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() &&
327	12	($this->handshakeMessage->capability & \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_DEPRECATE_EOF) !== 0
328		)
329		);
330
331		switch(true) {
332	12	case ($firstChar === \Plasma\Drivers\MySQL\Messages\ErrResponseMessage::getID()):
333	2	$message = new \Plasma\Drivers\MySQL\Messages\ErrResponseMessage($this);
334	2	break;
335	10	case ($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\StatementPrepareCommand && $firstChar === $okRespID):
336	2	$message = new \Plasma\Drivers\MySQL\Messages\PrepareStatementOkMessage($this);
337	2	break;
338	10	case $isOkMessage:
339	10	$message = new \Plasma\Drivers\MySQL\Messages\OkResponseMessage($this);
340	10	break;
341	4	case ($firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() && $length < 6):
342		$message = new \Plasma\Drivers\MySQL\Messages\EOFMessage($this);
343		break;
344		default:
345	4	$buffer->prepend($firstChar);
346
347	4	if($this->parseCallback !== null) {
348		$parse = $this->parseCallback;
349		$this->parseCallback = null;
350
351		$caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
352		$parse($caller);
353	4	} elseif($this->currentCommand !== null) {
354	4	$caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
355	4	$this->currentCommand->onNext($caller);
356		}
357
358		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Left over buffer: '.$buffer->getSize()) \|\| true));
359
360	4	if($this->buffer->getSize() > 0) {
361		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Scheduling future read with '.$this->buffer->getSize().' bytes') \|\| true));
362
363		$this->driver->getLoop()->futureTick(function () {
364	4	$this->processBuffer();
365	4	});
366		}
367
368	4	return;
369		break;
370		}
371		}
372
373		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Message '.\get_class($message)) \|\| true));
374
375	12	$state = $message->setParserState();
376	12	if($state !== -1) {
377	12	$this->state = $state;
378		}
379
380	12	if($message instanceof \Plasma\Drivers\MySQL\Messages\HandshakeMessage) {
381	12	$this->handshakeMessage = $message;
382		}
383
384	12	$this->handleMessage($buffer, $message);
385	12	}
386
387		/**
388		* Handles an incoming message.
389		* @param \Plasma\BinaryBuffer $buffer
390		* @param \Plasma\Drivers\MySQL\Messages\MessageInterface $message
391		* @return void
392		*/
393	12	function handleMessage(\Plasma\BinaryBuffer $buffer, \Plasma\Drivers\MySQL\Messages\MessageInterface $message) {
394		try {
395	12	$buffer = $message->parseMessage($buffer);
396	12	if(!$buffer) {
397		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned handle (unsufficent buffer length)') \|\| true));
398		return;
399		}
400
401	12	if($this->currentCommand !== null) {
402		if(
403	12	($message instanceof \Plasma\Drivers\MySQL\Messages\OkResponseMessage \|\| $message instanceof \Plasma\Drivers\MySQL\Messages\EOFMessage)
404	12	&& $this->currentCommand->hasFinished()
405		) {
406	10	$command = $this->currentCommand;
407	10	$this->currentCommand = null;
408
409	10	$command->onComplete();
410	6	} elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
411	2	$error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
412
413	2	$command = $this->currentCommand;
414	2	$this->currentCommand = null;
415
416	2	$command->onError($error);
417		} else {
418	4	$command = $this->currentCommand;
419	4	$command->onNext($message);
420
421	4	if($command->hasFinished()) {
422	2	if($this->currentCommand === $command) {
423		$this->currentCommand = null;
424		}
425
426	12	$command->onComplete();
427		}
428		}
429	12	} elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
430		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Error Response Message with message: '.$message->errorMessage) \|\| true));
431
432		$error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
433		$this->emit('error', array($error));
434		}
435
436	12	$this->emit('message', array($message));
437		} catch (\Plasma\Drivers\MySQL\Messages\ParseException $e) {
438		$state = $e->getState();
439		if($state !== null) {
440		$this->state = $state;
441		}
442
443		$buffer = $e->getBuffer();
444		if($buffer !== null) {
445		$this->buffer->clear();
446		$this->buffer->append($buffer);
447		}
448
449		if($this->currentCommand !== null) {
450		$this->currentCommand->onError($e);
451		}
452
453		$this->emit('error', array($e));
454		$this->connection->close();
455		}
456
457	12	if($this->buffer->getSize() > 0) {
458		// for debugging - \assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Scheduling future read (msg) with '.$this->buffer->getSize().' bytes') \|\| true));
459
460		$this->driver->getLoop()->futureTick(function () {
461	2	$this->processBuffer();
462	2	});
463		}
464	12	}
465
466		/**
467		* Adds the events to the connection.
468		* @return void
469		*/
470	12	protected function addEvents() {
471		$this->connection->on('data', function ($chunk) {
472	12	$this->buffer->append($chunk);
473	12	$this->processBuffer();
474	12	});
475
476		$this->connection->on('close', function () {
477	4	$this->handleClose();
478	12	});
479	12	}
480
481		/**
482		* Connection close handler.
483		* @return void
484		*/
485	4	protected function handleClose() {
486	4	if($this->state === static::STATE_AUTH \|\| $this->state === static::STATE_AUTH_SENT) {
487		$this->state = static::STATE_AUTH_ERROR;
488		}
489	4	}
490		}
491

PlasmaPHP / driver-mysql

GitHub Access Token became invalid

Push — master ( 396319...5c4ceb )

ProtocolParser::getHandshakeMessage() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like