diff --git a/.changes/nextrelease/fix-eventstream-partial-checksum-read.json b/.changes/nextrelease/fix-eventstream-partial-checksum-read.json new file mode 100644 index 0000000000..a3f672ed00 --- /dev/null +++ b/.changes/nextrelease/fix-eventstream-partial-checksum-read.json @@ -0,0 +1,7 @@ +[ + { + "type": "bugfix", + "category": "", + "description": "Fixed an issue in NonSeekableStreamDecodingEventStreamIterator where a message checksum returned across multiple partial reads caused a checksum mismatch." + } +] diff --git a/src/Api/Parser/NonSeekableStreamDecodingEventStreamIterator.php b/src/Api/Parser/NonSeekableStreamDecodingEventStreamIterator.php index d59f200e4a..20df0251cd 100644 --- a/src/Api/Parser/NonSeekableStreamDecodingEventStreamIterator.php +++ b/src/Api/Parser/NonSeekableStreamDecodingEventStreamIterator.php @@ -50,7 +50,21 @@ protected function parseEvent(): array ) ); $calculatedCrc = hash_final($this->hashContext, true); - $messageCrc = $this->stream->read(4); + $messageCrc = ''; + while ( + strlen($messageCrc) < self::BYTES_TRAILING + && !$this->stream->eof() + ) { + $chunk = $this->stream->read( + self::BYTES_TRAILING - strlen($messageCrc) + ); + if ($chunk === '') { + break; + } + + $messageCrc .= $chunk; + } + if ($calculatedCrc !== $messageCrc) { throw new ParserException('Message checksum mismatch.'); } diff --git a/tests/Api/Parser/NonSeekableStreamDecodingEventStreamIteratorTest.php b/tests/Api/Parser/NonSeekableStreamDecodingEventStreamIteratorTest.php index 9da889679f..8f8be32615 100644 --- a/tests/Api/Parser/NonSeekableStreamDecodingEventStreamIteratorTest.php +++ b/tests/Api/Parser/NonSeekableStreamDecodingEventStreamIteratorTest.php @@ -119,6 +119,40 @@ public function testReadAndHashBytesHandlesPartialReads( ); } + #[DataProvider('readAndHashBytesHandlesPartialReadsProvider')] + public function testParseEventHandlesPartialChecksumReads( + string $eventName, + array $expected + ): void + { + $eventPath = self::EVENT_STREAMS_DIR . "events/$eventName"; + $eventStream = Utils::streamFor( + base64_decode( + file_get_contents($eventPath) + ) + ); + $partialReadStream = $this->createMock(StreamInterface::class); + $partialReadStream->method('isSeekable')->willReturn(false); + $partialReadStream->method('isReadable')->willReturn(true); + $partialReadStream->method('read') + ->willReturnCallback(function ($length) use ($eventStream) { + return $eventStream->read(min($length, 1)); + }); + $iterator = new NonSeekableStreamDecodingEventStreamIterator( + $partialReadStream + ); + $iterator->rewind(); + $event = $iterator->current(); + $this->assertEquals( + $expected['headers'], + $event['headers'] + ); + $this->assertEquals( + $expected['payload'], + json_decode($event['payload']->getContents(), true) + ); + } + public static function readAndHashBytesHandlesPartialReadsProvider(): \Generator { $cases = json_decode(