nsq/nsqphp
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Snappy fix receiving partial message

Browse Source
This commit is contained in:
Konstantin Grachev
2021-02-01 01:17:39 +03:00
parent 3d8f5be2d0
commit 801fdfe547
1 changed files with 22 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/Socket/SnappySocket.php
View File

@@ -84,44 +84,50 @@ final class SnappySocket implements Socket
$size = $chunkType >> 8; $size = $chunkType >> 8;
$chunkType &= 0xff; $chunkType &= 0xff;
$this->logger->debug('Snappy receive chunk [{chunk}] with size [{size}]', [ $this->logger->debug('Snappy receive chunk [{chunk}], size [{size}]', [
'chunk' => $chunkType, 'chunk' => $chunkType,
'size' => $size, 'size' => $size,
]); ]);
do {
$input->append(
$this->socket->read($size),
);
$size -= $input->size();
} while ($input->size() < $size);
switch ($chunkType) { switch ($chunkType) {
case 0xff: case 0xff:
$this->logger->debug('Snappy identifier chunk'); $this->logger->debug('Snappy identifier chunk');
$this->socket->read(6); // discard identifier body $input->discard(6); // discard identifier body
break; break;
case 0x00: // 'compressed', case 0x00: // 'compressed',
$this->logger->debug('Snappy compressed chunk'); $this->logger->debug('Snappy compressed chunk');
$input->append( $data = $input
$this->socket->read($size),
)
->discard(4) // discard checksum ->discard(4) // discard checksum
->flush()
; ;
$output->append( $this->logger->debug('Snappy compressed data [{data}]', ['data' => $data]);
snappy_uncompress(
$input->flush(), $output->append(snappy_uncompress($data));
),
);
break; break;
case 0x01: // 'uncompressed', case 0x01: // 'uncompressed',
$this->logger->debug('Snappy uncompressed chunk'); $this->logger->debug('Snappy uncompressed chunk');
$input->append( $data = $input
$this->socket->read($size),
)
->discard(4) // discard checksum ->discard(4) // discard checksum
->flush()
; ;
$output->append($input->flush()); $this->logger->debug('Snappy uncompressed data [{data}]', ['data' => $data]);
$output->append($data);
break; break;
case 0xfe:// 'padding', case 0xfe:// 'padding',