ProtocolParser::setParseCallback() - Code Metrics - Inspection of "65ffdbdfbca16f4f7a16cf3ad27a35c9f8e8fb55" - PlasmaPHP/driver-mysql - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — master (65ffdb)

by Charlotte

created 2018-12-30 15:35 UTC

ProtocolParser::setParseCallback() A

↳ Parent: ProtocolParser

Complexity

Conditions	1
Paths	1

Size

Total Lines	2
Code Lines	1

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

Metric	Value
cc	1
eloc	1
nc	1
nop	1
dl	0
loc	2
ccs	0
cts	2
cp	0
crap	2
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_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 = 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\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->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;
    }
    
    /**
     * 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));
            
            $this->connection->write($length.$sequence.$partial);
        } 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));
            $this->connection->write($length.$sequence);
        }
    }
    
    /**
     * 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;
        }
        
        //\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Processing command '.get_class($command)) || true));
        
        $this->sendPacket($command->getEncodedMessage());
        
        if($command !== $this->currentCommand || !$command->waitForCompletion()) {
            //\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() {
        //\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('ProtocolParser::processBuffer called') || true));
        
        if($this->buffer->getSize() < 4) {
            //\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();
        
        //\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('First 10 bytes: '.implode(', ', unpack('C*', \substr($this->buffer->getContents(), 0, 10)))) || true));
        //\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Read packet header length ('.$length.') and sequence ('.$this->sequenceID.')') || true));
        
        if($length === static::CLIENT_MAX_PACKET_SIZE) {
            $this->buffer->read(($length + 4));
            $this->messageBuffer->append($buffer->read($length));
            
            //\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) {
            //\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) {
            //\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);
            //\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\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 ($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) {
                        $command = $this->currentCommand;
                        
                        $caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
                        $command->onNext($caller);
                        
                        if($command->hasFinished()) {
                            $this->currentCommand = null;
                            $command->onComplete();
                        }
                    }
                    
                    //\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Left over buffer: '.$buffer->getSize()) || true));
                    
                    if($this->buffer->getSize() > 0) {
                        //\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;
            }
        }
        
        //\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) {
                //\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) {
                //\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) {
            //\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_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 = 0x1000000;
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 = 0;
111
112		/**
113		* @var \Plasma\Drivers\MySQL\Messages\HandshakeMessage\|null
114		*/
115		protected $handshakeMessage;
116
117		/**
118		* @var \Plasma\Drivers\MySQL\Messages\OkResponseMessage\|null
119		*/
120		protected $lastOkMessage;
121
122		/**
123		* @var \Plasma\CommandInterface\|null
124		*/
125		protected $currentCommand;
126
127		/**
128		* @var callable\|null
129		*/
130		protected $parseCallback;
131
132		/**
133		* Constructor.
134		* @param \Plasma\Drivers\MySQL\Driver $driver
135		* @param \React\Socket\ConnectionInterface $connection
136		*/
137	57	function __construct(\Plasma\Drivers\MySQL\Driver $driver, \React\Socket\ConnectionInterface $connection) {
138	57	$this->driver = $driver;
139	57	$this->connection = $connection;
140
141	57	$this->buffer = new \Plasma\BinaryBuffer();
142	57	$this->messageBuffer = new \Plasma\BinaryBuffer();
143
144	57	$this->addEvents();
145	57	}
146
147		/**
148		* Invoke a command to execute.
149		* @param \Plasma\CommandInterface\|null $command
150		* @return void
151		*/
152	56	function invokeCommand(?\Plasma\CommandInterface $command): void {
153	56	if($command === null) {
154	53	return;
155		}
156
157	56	$this->currentCommand = $command;
158	56	$this->processCommand();
159	56	}
160
161		/**
162		* Executes a command, without handling any aftermath.
163		* The `onComplete` callback will be immediately invoked, regardless of the `waitForCompletion` value.
164		* @param \Plasma\CommandInterface $command
165		* @return void
166		*/
167		function executeCommand(\Plasma\CommandInterface $command): void {
168		$this->processCommand($command);
169		}
170
171		/**
172		* Marks the command itself as finished, if currently running.
173		* @param \Plasma\Drivers\MySQL\Commands\CommandInterface $command
174		* @return void
175		*/
176	53	function markCommandAsFinished(\Plasma\CommandInterface $command): void {
177	53	if($command === $this->currentCommand) {
178	53	$this->currentCommand = null;
179		}
180
181	53	$command->onComplete();
182	53	}
183
184		/**
185		* Get the parser state.
186		* @return int
187		*/
188	1	function getState(): int {
189	1	return $this->state;
190		}
191
192		/**
193		* Get the handshake message, or null.
194		* @return \Plasma\Drivers\MySQL\Messages\HandshakeMessage\|null
195		*/
196	56	function getHandshakeMessage(): ?\Plasma\Drivers\MySQL\Messages\HandshakeMessage {
197	56	return $this->handshakeMessage;
198		}
199
200		/**
201		* Get the last ok response message, or null.
202		* @return \Plasma\Drivers\MySQL\Messages\OkResponseMessage\|null
203		*/
204	4	function getLastOkMessage(): ?\Plasma\Drivers\MySQL\Messages\OkResponseMessage {
205	4	return $this->lastOkMessage;
206		}
207
208		/**
209		* Sends a packet to the server.
210		* @param string $packet
211		* @return void
212		*/
213	57	function sendPacket(string $packet): void {
214	57	$initPacklen = \strlen($packet);
215
216		do {
217	57	$partial = \substr($packet, 0, static::CLIENT_MAX_PACKET_SIZE);
218	57	$partlen = \strlen($partial);
219
220	57	$packet = \substr($packet, static::CLIENT_MAX_PACKET_SIZE);
221	57	$packlen = \strlen($packet);
222
223	57	$length = \Plasma\BinaryBuffer::writeInt3($partlen);
224	57	$sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
225
226	57	$this->connection->write($length.$sequence.$partial);
227	57	} while($packlen > static::CLIENT_MAX_PACKET_SIZE);
228
229		// If the packet is exactly the max size, we have to send two packets
230	57	if($initPacklen === static::CLIENT_MAX_PACKET_SIZE) {
231	1	$length = \Plasma\BinaryBuffer::writeInt3(0);
232	1	$sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
233	1	$this->connection->write($length.$sequence);
234		}
235	57	}
236
237		/**
238		* Sets the parse callback.
239		* @param callable $callback
240		* @return void
241		*/
242		function setParseCallback(callable $callback): void {
243		$this->parseCallback = $callback;
244		}
245
246		/**
247		* Processes a command.
248		* @param \Plasma\CommandInterface\|null $command
249		* @return void
250		*/
251	56	protected function processCommand(?\Plasma\CommandInterface $command = null) {
252	56	if($command === null && $this->currentCommand instanceof \Plasma\CommandInterface) {
253	56	$command = $this->currentCommand;
254
255	56	if($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) {
256	56	$state = $command->setParserState();
257	56	if($state !== -1) {
258	56	$this->state = $state;
259		}
260		}
261		}
262
263	56	if($command === null) {
264		return;
265		}
266
267	56	if(!($command instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) \|\| $command->resetSequence()) {
268	53	$this->sequenceID = -1;
269		}
270
271		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Processing command '.get_class($command)) \|\| true));
272
273	56	$this->sendPacket($command->getEncodedMessage());
274
275	56	if($command !== $this->currentCommand \|\| !$command->waitForCompletion()) {
276		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Mark command as completed') \|\| true));
277	35	$command->onComplete();
278
279	35	if($command === $this->currentCommand) {
280	35	$this->currentCommand = null;
281		}
282		}
283	56	}
284
285		/**
286		* Processes the buffer.
287		* @return void
288		*/
289	56	protected function processBuffer() {
290		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('ProtocolParser::processBuffer called') \|\| true));
291
292	56	if($this->buffer->getSize() < 4) {
293		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Not enough data received for packet header ('.$this->buffer->getSize().')') \|\| true));
294		return;
295		}
296
297	56	$buffer = clone $this->buffer;
298
299	56	$length = $buffer->readInt3();
300	56	$this->sequenceID = $buffer->readInt1();
301
302		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('First 10 bytes: '.implode(', ', unpack('C*', \substr($this->buffer->getContents(), 0, 10)))) \|\| true));
303		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Read packet header length ('.$length.') and sequence ('.$this->sequenceID.')') \|\| true));
304
305	56	if($length === static::CLIENT_MAX_PACKET_SIZE) {
306		$this->buffer->read(($length + 4));
307		$this->messageBuffer->append($buffer->read($length));
308
309		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned, 16mb packet received, waiting for the last one to arrive') \|\| true));
310		return;
311	56	} elseif($this->messageBuffer->getSize() > 0) {
312		$this->messageBuffer->append($buffer->read($length));
313		$buffer = $this->messageBuffer;
314		$this->messageBuffer = new \Plasma\BinaryBuffer();
315		}
316
317	56	if($buffer->getSize() < $length) {
318		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned, insufficent length: '.$buffer->getSize().', '.$length.' required') \|\| true));
319		return;
320		}
321
322	56	if($length > 0) {
323	56	$this->buffer->read(($length + 4));
324	56	$buffer->slice(0, $length);
325		} else {
326		$this->buffer->slice($buffer->getSize());
327		}
328
329	56	if($buffer->getSize() === 0) {
330		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Buffer length is 0') \|\| true));
331		return;
332		}
333
334		/** @var \Plasma\Drivers\MySQL\Messages\MessageInterface $message */
335	56	$message = null;
336
337	56	if($this->state === static::STATE_INIT) {
338	56	$message = new \Plasma\Drivers\MySQL\Messages\HandshakeMessage($this);
339		} else {
340	56	$firstChar = $buffer->read(1);
341		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Message char "'.$firstChar.'" (0x'.\dechex(\ord($firstChar)).') - buffer length: '.$buffer->getSize()) \|\| true));
342
343	56	$okRespID = \Plasma\Drivers\MySQL\Messages\OkResponseMessage::getID();
344		$isOkMessage = (
345		(
346	56	$firstChar === $okRespID &&
347	55	(!($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\QueryCommand)
348	55	\|\| \strtoupper(\substr($this->currentCommand->getQuery(), 0, 6)) !== 'SELECT') // Fix for MySQL 5.7
349		) \|\|
350		(
351	44	$firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() &&
352	56	($this->handshakeMessage->capability & \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_DEPRECATE_EOF) !== 0
353		)
354		);
355
356		switch(true) {
357	56	case ($firstChar === \Plasma\Drivers\MySQL\Messages\ErrResponseMessage::getID()):
358	1	$message = new \Plasma\Drivers\MySQL\Messages\ErrResponseMessage($this);
359	1	break;
360	55	case ($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\StatementPrepareCommand && $firstChar === $okRespID):
361	36	$message = new \Plasma\Drivers\MySQL\Messages\PrepareStatementOkMessage($this);
362	36	break;
363	55	case $isOkMessage:
364	55	$message = new \Plasma\Drivers\MySQL\Messages\OkResponseMessage($this);
365	55	$this->lastOkMessage = $message;
366	55	break;
367	43	case ($firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() && $length < 6):
368		$message = new \Plasma\Drivers\MySQL\Messages\EOFMessage($this);
369		break;
370		default:
371	43	$buffer->prepend($firstChar);
372
373	43	if($this->parseCallback !== null) {
374		$parse = $this->parseCallback;
375		$this->parseCallback = null;
376
377		$caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
378		$parse($caller);
379	43	} elseif($this->currentCommand !== null) {
380	43	$command = $this->currentCommand;
381
382	43	$caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
383	43	$command->onNext($caller);
384
385	43	if($command->hasFinished()) {
386	36	$this->currentCommand = null;
387	36	$command->onComplete();
388		}
389		}
390
391		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Left over buffer: '.$buffer->getSize()) \|\| true));
392
393	43	if($this->buffer->getSize() > 0) {
394		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Scheduling future read with '.$this->buffer->getSize().' bytes') \|\| true));
395
396		$this->driver->getLoop()->futureTick(function () {
397	43	$this->processBuffer();
398	43	});
399		}
400
401	43	return;
402		break;
403		}
404		}
405
406		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Message '.\get_class($message)) \|\| true));
407
408	56	$state = $message->setParserState();
409	56	if($state !== -1) {
410	56	$this->state = $state;
411		}
412
413	56	if($message instanceof \Plasma\Drivers\MySQL\Messages\HandshakeMessage) {
414	56	$this->handshakeMessage = $message;
415		}
416
417	56	$this->handleMessage($buffer, $message);
418	56	}
419
420		/**
421		* Handles an incoming message.
422		* @param \Plasma\BinaryBuffer $buffer
423		* @param \Plasma\Drivers\MySQL\Messages\MessageInterface $message
424		* @return void
425		*/
426	56	function handleMessage(\Plasma\BinaryBuffer $buffer, \Plasma\Drivers\MySQL\Messages\MessageInterface $message) {
427		try {
428	56	$buffer = $message->parseMessage($buffer);
429	56	if(!$buffer) {
430		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned handle (unsufficent buffer length)') \|\| true));
431		return;
432		}
433
434	56	if($this->currentCommand !== null) {
435		if(
436	56	($message instanceof \Plasma\Drivers\MySQL\Messages\OkResponseMessage \|\| $message instanceof \Plasma\Drivers\MySQL\Messages\EOFMessage)
437	56	&& $this->currentCommand->hasFinished()
438		) {
439	55	$command = $this->currentCommand;
440	55	$this->currentCommand = null;
441
442	55	$command->onComplete();
443	54	} elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
444	1	$error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
445
446	1	$command = $this->currentCommand;
447	1	$this->currentCommand = null;
448
449	1	$command->onError($error);
450		} else {
451	53	$command = $this->currentCommand;
452	53	$command->onNext($message);
453
454	53	if($command->hasFinished()) {
455	53	if($this->currentCommand === $command) {
456		$this->currentCommand = null;
457		}
458
459	56	$command->onComplete();
460		}
461		}
462	56	} elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
463		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Error Response Message with message: '.$message->errorMessage) \|\| true));
464
465		$error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
466		$this->emit('error', array($error));
467		}
468
469	56	$this->emit('message', array($message));
470		} catch (\Plasma\Drivers\MySQL\Messages\ParseException $e) {
471		$state = $e->getState();
472		if($state !== null) {
473		$this->state = $state;
474		}
475
476		$buffer = $e->getBuffer();
477		if($buffer !== null) {
478		$this->buffer->clear();
479		$this->buffer->append($buffer);
480		}
481
482		if($this->currentCommand !== null) {
483		$this->currentCommand->onError($e);
484		}
485
486		$this->emit('error', array($e));
487		$this->connection->close();
488		}
489
490	56	if($this->buffer->getSize() > 0) {
491		//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Scheduling future read (msg) with '.$this->buffer->getSize().' bytes') \|\| true));
492
493		$this->driver->getLoop()->futureTick(function () {
494	36	$this->processBuffer();
495	36	});
496		}
497	56	}
498
499		/**
500		* Adds the events to the connection.
501		* @return void
502		*/
503	57	protected function addEvents() {
504		$this->connection->on('data', function ($chunk) {
505	56	$this->buffer->append($chunk);
506	56	$this->processBuffer();
507	57	});
508
509		$this->connection->on('close', function () {
510	3	$this->handleClose();
511	57	});
512	57	}
513
514		/**
515		* Connection close handler.
516		* @return void
517		*/
518	3	protected function handleClose() {
519	3	if($this->state === static::STATE_AUTH \|\| $this->state === static::STATE_AUTH_SENT) {
520		$this->state = static::STATE_AUTH_ERROR;
521		}
522	3	}
523		}
524

PlasmaPHP / driver-mysql

GitHub Access Token became invalid

Branch — master (65ffdb)

ProtocolParser::setParseCallback() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like