ProtocolParser::processCommand() - Code Metrics - Inspection of "Set SQL mode for type tests" - PlasmaPHP/driver-mysql - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 1bb41b...8abffd )

by Charlotte

created 2018-11-23 20:09 UTC

ProtocolParser::processCommand() B

↳ Parent: ProtocolParser

Complexity

Conditions	11
Paths	28

Size

Total Lines	30
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	11
eloc	15
nc	28
nop	1
dl	0
loc	30
rs	7.3166
c	0
b	0
f	0

How to fix Complexity

<?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
 * @codeCoverageIgnore
 */
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\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 {
        $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;
        }
        
        //\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 === 0xFFFFFF) {
            $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\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);
                    $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;
        }
    }
}


GitHub Access Token became invalid

Push — master ( 1bb41b...8abffd )

ProtocolParser::processCommand() B

Complexity

Size

Duplication

Importance

How to fix Complexity

Long Method

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			* @codeCoverageIgnore
16			*/
17			class ProtocolParser implements \Evenement\EventEmitterInterface {
18			use \Evenement\EventEmitterTrait;
19
20			/**
21			* @var int
22			*/
23			const STATE_INIT = 0;
24
25			/**
26			* @var int
27			*/
28			const STATE_HANDSHAKE = 1;
29
30			/**
31			* @var int
32			*/
33			const STATE_HANDSHAKE_ERROR = 2;
34
35			/**
36			* @var int
37			*/
38			const STATE_AUTH = 5;
39
40			/**
41			* @var int
42			*/
43			const STATE_AUTH_SENT = 6;
44
45			/**
46			* @var int
47			*/
48			const STATE_AUTH_ERROR = 7;
49
50			/**
51			* @var int
52			*/
53			const STATE_OK = 9;
54
55			/**
56			* @var int
57			*/
58			const CLIENT_CAPABILITIES = (
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			function __construct(\Plasma\Drivers\MySQL\Driver $driver, \React\Socket\ConnectionInterface $connection) {
138			$this->driver = $driver;
139			$this->connection = $connection;
140
141			$this->buffer = new \Plasma\BinaryBuffer();
142			$this->messageBuffer = new \Plasma\BinaryBuffer();
143
144			$this->addEvents();
145			}
146
147			/**
148			* Invoke a command to execute.
149			* @param \Plasma\CommandInterface\|null $command
150			* @return void
151			*/
152			function invokeCommand(?\Plasma\CommandInterface $command): void {
153			if($command === null) {
154			return;
155			}
156
157			$this->currentCommand = $command;
158			$this->processCommand();
159			}
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			function markCommandAsFinished(\Plasma\CommandInterface $command): void {
177			if($command === $this->currentCommand) {
178			$this->currentCommand = null;
179			}
180
181			$command->onComplete();
182			}
183
184			/**
185			* Get the parser state.
186			* @return int
187			*/
188			function getState(): int {
189			return $this->state;
190			}
191
192			/**
193			* Get the handshake message, or null.
194			* @return \Plasma\Drivers\MySQL\Messages\HandshakeMessage\|null
195			*/
196			function getHandshakeMessage(): ?\Plasma\Drivers\MySQL\Messages\HandshakeMessage {
197			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			function getLastOkMessage(): ?\Plasma\Drivers\MySQL\Messages\OkResponseMessage {
205			return $this->lastOkMessage;
206			}
207
208			/**
209			* Sends a packet to the server.
210			* @param string $packet
211			* @return void
212			*/
213			function sendPacket(string $packet): void {
214			$maxSize = static::CLIENT_MAX_PACKET_SIZE - 4;
215
216			do {
217			$partial = \substr($packet, 0, $maxSize);
218			$packet = \substr($packet, $maxSize);
219
220			$length = \Plasma\BinaryBuffer::writeInt3(\strlen($partial));
221			$sequence = \Plasma\BinaryBuffer::writeInt1((++$this->sequenceID));
222
223			$this->connection->write($length.$sequence.$partial);
224			} while(\strlen($packet) > $maxSize);
225			}
226
227			/**
228			* Sets the parse callback.
229			* @param callable $callback
230			* @return void
231			*/
232			function setParseCallback(callable $callback): void {
233			$this->parseCallback = $callback;
234			}
235
236			/**
237			* Processes a command.
238			* @param \Plasma\CommandInterface\|null $command
239			* @return void
240			*/
241			protected function processCommand(?\Plasma\CommandInterface $command = null) {
242			if($command === null && $this->currentCommand instanceof \Plasma\CommandInterface) {
243			$command = $this->currentCommand;
244
245			if($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) {
246			$state = $command->setParserState();
247			if($state !== -1) {
248			$this->state = $state;
249			}
250			}
251			}
252
253			if($command === null) {
254			return;
255			}
256
257			if(!($command instanceof \Plasma\Drivers\MySQL\Commands\CommandInterface) \|\| $command->resetSequence()) {
258			$this->sequenceID = -1;
259			}
260
261			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Processing command '.get_class($command)) \|\| true));
262
263			$this->sendPacket($command->getEncodedMessage());
264
265			if($command !== $this->currentCommand \|\| !$command->waitForCompletion()) {
266			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Mark command as completed') \|\| true));
267			$command->onComplete();
268
269			if($command === $this->currentCommand) {
270			$this->currentCommand = null;
271			}
272			}
273			}
274
275			/**
276			* Processes the buffer.
277			* @return void
278			*/
279			protected function processBuffer() {
280			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('ProtocolParser::processBuffer called') \|\| true));
281
282			if($this->buffer->getSize() < 4) {
283			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Not enough data received for packet header ('.$this->buffer->getSize().')') \|\| true));
284			return;
285			}
286
287			$buffer = clone $this->buffer;
288
289			$length = $buffer->readInt3();
290			$this->sequenceID = $buffer->readInt1();
291
292			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('First 10 bytes: '.implode(', ', unpack('C*', \substr($this->buffer->getContents(), 0, 10)))) \|\| true));
293			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Read packet header length ('.$length.') and sequence ('.$this->sequenceID.')') \|\| true));
294
295			if($length === 0xFFFFFF) {
296			$this->buffer->read(($length + 4));
297			$this->messageBuffer->append($buffer->read($length));
298
299			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned, 16mb packet received, waiting for the last one to arrive') \|\| true));
300			return;
301			} elseif($this->messageBuffer->getSize() > 0) {
302			$this->messageBuffer->append($buffer->read($length));
303			$buffer = $this->messageBuffer;
304			$this->messageBuffer = new \Plasma\BinaryBuffer();
305			}
306
307			if($buffer->getSize() < $length) {
308			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned, insufficent length: '.$buffer->getSize().', '.$length.' required') \|\| true));
309			return;
310			}
311
312			if($length > 0) {
313			$this->buffer->read(($length + 4));
314			$buffer->slice(0, $length);
315			} else {
316			$this->buffer->slice($buffer->getSize());
317			}
318
319			if($buffer->getSize() === 0) {
320			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Buffer length is 0') \|\| true));
321			return;
322			}
323
324			/** @var \Plasma\Drivers\MySQL\Messages\MessageInterface $message */
325			$message = null;
326
327			if($this->state === static::STATE_INIT) {
328			$message = new \Plasma\Drivers\MySQL\Messages\HandshakeMessage($this);
329			} else {
330			$firstChar = $buffer->read(1);
331			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Message char "'.$firstChar.'" (0x'.\dechex(\ord($firstChar)).') - buffer length: '.$buffer->getSize()) \|\| true));
332
333			$okRespID = \Plasma\Drivers\MySQL\Messages\OkResponseMessage::getID();
334			$isOkMessage = (
335			(
336			$firstChar === $okRespID &&
337			!($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\StatementExecuteCommand)
338			) \|\|
339			(
340			$firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() &&
341			($this->handshakeMessage->capability & \Plasma\Drivers\MySQL\CapabilityFlags::CLIENT_DEPRECATE_EOF) !== 0
342			)
343			);
344
345			switch(true) {
346			case ($firstChar === \Plasma\Drivers\MySQL\Messages\ErrResponseMessage::getID()):
347			$message = new \Plasma\Drivers\MySQL\Messages\ErrResponseMessage($this);
348			break;
349			case ($this->currentCommand instanceof \Plasma\Drivers\MySQL\Commands\StatementPrepareCommand && $firstChar === $okRespID):
350			$message = new \Plasma\Drivers\MySQL\Messages\PrepareStatementOkMessage($this);
351			break;
352			case $isOkMessage:
353			$message = new \Plasma\Drivers\MySQL\Messages\OkResponseMessage($this);
354			$this->lastOkMessage = $message;
355			break;
356			case ($firstChar === \Plasma\Drivers\MySQL\Messages\EOFMessage::getID() && $length < 6):
357			$message = new \Plasma\Drivers\MySQL\Messages\EOFMessage($this);
358			break;
359			default:
360			$buffer->prepend($firstChar);
361
362			if($this->parseCallback !== null) {
363			$parse = $this->parseCallback;
364			$this->parseCallback = null;
365
366			$caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
367			$parse($caller);
368			} elseif($this->currentCommand !== null) {
369			$command = $this->currentCommand;
370
371			$caller = new \Plasma\Drivers\MySQL\ProtocolOnNextCaller($this, $buffer);
372			$command->onNext($caller);
373
374			if($command->hasFinished()) {
375			$this->currentCommand = null;
376			$command->onComplete();
377			}
378			}
379
380			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Left over buffer: '.$buffer->getSize()) \|\| true));
381
382			if($this->buffer->getSize() > 0) {
383			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Scheduling future read with '.$this->buffer->getSize().' bytes') \|\| true));
384
385			$this->driver->getLoop()->futureTick(function () {
386			$this->processBuffer();
387			});
388			}
389
390			return;
391			break;
392			}
393			}
394
395			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Message '.\get_class($message)) \|\| true));
396
397			$state = $message->setParserState();
398			if($state !== -1) {
399			$this->state = $state;
400			}
401
402			if($message instanceof \Plasma\Drivers\MySQL\Messages\HandshakeMessage) {
403			$this->handshakeMessage = $message;
404			}
405
406			$this->handleMessage($buffer, $message);
407			}
408
409			/**
410			* Handles an incoming message.
411			* @param \Plasma\BinaryBuffer $buffer
412			* @param \Plasma\Drivers\MySQL\Messages\MessageInterface $message
413			* @return void
414			*/
415			function handleMessage(\Plasma\BinaryBuffer $buffer, \Plasma\Drivers\MySQL\Messages\MessageInterface $message) {
416			try {
417			$buffer = $message->parseMessage($buffer);
418			if(!$buffer) {
419			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('returned handle (unsufficent buffer length)') \|\| true));
420			return;
421			}
422
423			if($this->currentCommand !== null) {
424			if(
425			($message instanceof \Plasma\Drivers\MySQL\Messages\OkResponseMessage \|\| $message instanceof \Plasma\Drivers\MySQL\Messages\EOFMessage)
426			&& $this->currentCommand->hasFinished()
427			) {
428			$command = $this->currentCommand;
429			$this->currentCommand = null;
430
431			$command->onComplete();
432			} elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
433			$error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
434
435			$command = $this->currentCommand;
436			$this->currentCommand = null;
437
438			$command->onError($error);
439			} else {
440			$command = $this->currentCommand;
441			$command->onNext($message);
442
443			if($command->hasFinished()) {
444			if($this->currentCommand === $command) {
445			$this->currentCommand = null;
446			}
447
448			$command->onComplete();
449			}
450			}
451			} elseif($message instanceof \Plasma\Drivers\MySQL\Messages\ErrResponseMessage) {
452			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Received Error Response Message with message: '.$message->errorMessage) \|\| true));
453
454			$error = new \Plasma\Exception($message->errorMessage, $message->errorCode);
455			$this->emit('error', array($error));
456			}
457
458			$this->emit('message', array($message));
459			} catch (\Plasma\Drivers\MySQL\Messages\ParseException $e) {
460			$state = $e->getState();
461			if($state !== null) {
462			$this->state = $state;
463			}
464
465			$buffer = $e->getBuffer();
466			if($buffer !== null) {
467			$this->buffer->clear();
468			$this->buffer->append($buffer);
469			}
470
471			if($this->currentCommand !== null) {
472			$this->currentCommand->onError($e);
473			}
474
475			$this->emit('error', array($e));
476			$this->connection->close();
477			}
478
479			if($this->buffer->getSize() > 0) {
480			//\assert((\Plasma\Drivers\MySQL\Messages\MessageUtility::debug('Scheduling future read (msg) with '.$this->buffer->getSize().' bytes') \|\| true));
481
482			$this->driver->getLoop()->futureTick(function () {
483			$this->processBuffer();
484			});
485			}
486			}
487
488			/**
489			* Adds the events to the connection.
490			* @return void
491			*/
492			protected function addEvents() {
493			$this->connection->on('data', function ($chunk) {
494			$this->buffer->append($chunk);
495			$this->processBuffer();
496			});
497
498			$this->connection->on('close', function () {
499			$this->handleClose();
500			});
501			}
502
503			/**
504			* Connection close handler.
505			* @return void
506			*/
507			protected function handleClose() {
508			if($this->state === static::STATE_AUTH \|\| $this->state === static::STATE_AUTH_SENT) {
509			$this->state = static::STATE_AUTH_ERROR;
510			}
511			}
512			}
513

PlasmaPHP / driver-mysql

GitHub Access Token became invalid

Push — master ( 1bb41b...8abffd )

ProtocolParser::processCommand() B

Complexity

Size

Duplication

Importance

How to fix Complexity

Long Method

Duplication Side-by-Side

Filter issues like