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