From 03758ba2b3342574e10d365ae6f166cf3375e464 Mon Sep 17 00:00:00 2001 From: Belisoful Date: Thu, 9 Jul 2026 08:23:41 +0000 Subject: [PATCH] TBufferStream - first-in, first-out byte buffer --- framework/IO/Stream/TBufferStream.php | 258 +++++++++++++++++++++ framework/classes.php | 1 + tests/unit/IO/Stream/TBufferStreamTest.php | 240 +++++++++++++++++++ 3 files changed, 499 insertions(+) create mode 100644 framework/IO/Stream/TBufferStream.php create mode 100644 tests/unit/IO/Stream/TBufferStreamTest.php diff --git a/framework/IO/Stream/TBufferStream.php b/framework/IO/Stream/TBufferStream.php new file mode 100644 index 000000000..da1dd059e --- /dev/null +++ b/framework/IO/Stream/TBufferStream.php @@ -0,0 +1,258 @@ + + * @link https://github.com/pradosoft/prado + * @license https://github.com/pradosoft/prado/blob/master/LICENSE + */ + +namespace Prado\IO\Stream; + +use Prado\TComponent; +use Psr\Http\Message\StreamInterface; + +/** + * TBufferStream class. + * + * Holds a first-in, first-out byte buffer: {@see write()} appends to the back and + * {@see read()} drains from the front, removing what it returns. It is a non-seekable + * pipe between a producer and a consumer, useful for handing bytes from one to the other + * without a file or socket. + * + * {@see getSize()} reports the bytes currently buffered, and {@see eof()} is true while + * the buffer is empty. Reading past the buffered data returns ''. + * + * @author Brad Anderson + * @since 4.4.0 + */ +class TBufferStream extends TComponent implements StreamInterface +{ + /** @var int The minimum consumed-prefix size before {@see read()} compacts; it also waits until the consumed prefix outweighs the unread bytes. */ + public const CHUNK_SIZE = 8192; + + /** @var string The buffered bytes; the unread region starts at {@see $_bufferPos}. */ + private string $_buffer = ''; + + /** @var int The offset of the first unread byte in {@see $_buffer}. */ + private int $_bufferPos = 0; + + /** @var bool Whether the stream has been closed or detached, leaving it unusable. */ + private bool $_detached = false; + + // + // ─── Self-encapsulated raw accessors ───────────────────────────────────── + // + + /** + * Returns the unread bytes by reference, compacting away any consumed prefix first so a + * caller can mutate the buffer in place (`$b = &$this->getBufferDirect(); $b .= ...`). The + * reference is valid only until the next call that replaces the buffer ({@see read()} on + * compaction, {@see setBufferDirect()}), which rebinds the property. + * @return string The unread buffered bytes, by reference. + */ + protected function &getBufferDirect(): string + { + if ($this->_bufferPos > 0) { + $this->_buffer = substr($this->_buffer, $this->_bufferPos); + $this->_bufferPos = 0; + } + return $this->_buffer; + } + + /** + * Sets the raw buffered bytes, discarding any consumed prefix. + * @param string $value The raw buffered bytes. + */ + protected function setBufferDirect(string $value): void + { + $this->_buffer = $value; + $this->_bufferPos = 0; + } + + /** + * Returns all buffered bytes from the front without consuming them; {@see getContents()} drains + * the buffer. + * @return string The full buffered contents. + */ + public function __toString(): string + { + return $this->getBufferDirect(); + } + + /** + * Reads and removes up to $length bytes from the front of the buffer. + * @param int $length The maximum number of bytes to read. + * @return string The bytes read; '' when the buffer is empty or $length is not positive. + */ + public function read(int $length): string + { + if ($this->_detached) { + throw new \RuntimeException('Cannot read from a closed TBufferStream'); + } + // Advance a read offset instead of re-slicing the remainder each call (which is O(n) per + // read and quadratic over a large buffer drained in small reads); compact only when the + // consumed prefix dominates. + $pos = $this->_bufferPos; + $available = strlen($this->_buffer) - $pos; + if ($length <= 0 || $available <= 0) { + return ''; + } + $take = $length < $available ? $length : $available; + $data = substr($this->_buffer, $pos, $take); + $pos += $take; + if ($pos >= static::CHUNK_SIZE && $pos >= strlen($this->_buffer) - $pos) { + $this->_buffer = substr($this->_buffer, $pos); + $pos = 0; + } + $this->_bufferPos = $pos; + return $data; + } + + /** + * Appends bytes to the back of the buffer. + * @param string $string The bytes to append. + * @return int The number of bytes appended. + */ + public function write(string $string): int + { + if ($this->_detached) { + throw new \RuntimeException('Cannot write to a closed TBufferStream'); + } + $this->_buffer .= $string; + return strlen($string); + } + + /** + * Returns the remaining buffered bytes, draining the buffer. + * @return string The remaining contents. + */ + public function getContents(): string + { + if ($this->_detached) { + throw new \RuntimeException('Cannot read from a closed TBufferStream'); + } + $data = $this->getBufferDirect(); + $this->setBufferDirect(''); + return $data; + } + + /** + * Returns the number of bytes currently buffered. + * @return ?int The buffered byte count. + */ + public function getSize(): ?int + { + return $this->_detached ? null : strlen($this->_buffer) - $this->_bufferPos; + } + + /** + * Indicates whether the buffer is empty. + * @return bool True when no bytes are buffered. + */ + public function eof(): bool + { + return $this->_bufferPos >= strlen($this->_buffer); + } + + /** + * Detaches the stream, emptying the buffer and marking it unusable. A buffer has no underlying + * resource to return, so this returns null; a later read or write throws. + * @return null Always null; the buffer has no resource. + */ + public function detach() + { + $this->setBufferDirect(''); + $this->_detached = true; + return null; + } + + /** + * Closes the stream, emptying the buffer and marking it unusable; a later read or write throws. + */ + public function close(): void + { + $this->setBufferDirect(''); + $this->_detached = true; + } + + /** + * Empties the buffer and returns it to a fresh, usable state, clearing any closed or detached + * flag so the object can be reused. This is a buffer convenience outside PSR-7, distinct from + * {@see close()}, which is terminal. + * @return static This buffer. + */ + public function reset(): static + { + $this->setBufferDirect(''); + $this->_detached = false; + return $this; + } + + /** + * Indicates whether the buffer is writable: true until closed or detached. + * @return bool Whether the stream is writable. + */ + public function isWritable(): bool + { + return !$this->_detached; + } + + /** + * Indicates whether the buffer is readable: true until closed or detached. + * @return bool Whether the stream is readable. + */ + public function isReadable(): bool + { + return !$this->_detached; + } + + /** + * Indicates the buffer is not seekable. + * @return bool Always false. + */ + public function isSeekable(): bool + { + return false; + } + + /** + * Throws, as a FIFO buffer cannot report a position. + * @throws \RuntimeException Always. + */ + public function tell(): int + { + throw new \RuntimeException('A TBufferStream has no fixed position'); + } + + /** + * Throws, as a FIFO buffer cannot seek. + * @param int $offset The stream offset (unused). + * @param int $whence The whence (unused). + * @throws \RuntimeException Always. + */ + public function seek(int $offset, int $whence = SEEK_SET): void + { + throw new \RuntimeException('A TBufferStream cannot seek'); + } + + /** + * Throws, as a FIFO buffer cannot rewind. + * @throws \RuntimeException Always. + */ + public function rewind(): void + { + throw new \RuntimeException('A TBufferStream cannot rewind'); + } + + /** + * Returns stream metadata, of which a buffer has none. + * @param ?string $key A specific metadata key, or null for the whole array. + * @return mixed An empty array, or null for a specific key. + */ + public function getMetadata(?string $key = null): mixed + { + return $key === null ? [] : null; + } +} diff --git a/framework/classes.php b/framework/classes.php index dd47217ab..80bdc3181 100644 --- a/framework/classes.php +++ b/framework/classes.php @@ -292,6 +292,7 @@ 'TSocketStream' => 'Prado\IO\Socket\TSocketStream', 'IStreamDecoratorPooling' => 'Prado\IO\Stream\IStreamDecoratorPooling', 'TBinaryStream' => 'Prado\IO\Stream\TBinaryStream', +'TBufferStream' => 'Prado\IO\Stream\TBufferStream', 'TDeflateStream' => 'Prado\IO\Stream\TDeflateStream', 'TInflateStream' => 'Prado\IO\Stream\TInflateStream', 'TStreamDecorator' => 'Prado\IO\Stream\TStreamDecorator', diff --git a/tests/unit/IO/Stream/TBufferStreamTest.php b/tests/unit/IO/Stream/TBufferStreamTest.php new file mode 100644 index 000000000..ac11375b8 --- /dev/null +++ b/tests/unit/IO/Stream/TBufferStreamTest.php @@ -0,0 +1,240 @@ +write('AAA')); + self::assertSame(3, $b->write('BBB')); + self::assertSame(6, $b->getSize()); + self::assertSame('AAAB', $b->read(4), 'Reads drain from the front.'); + self::assertSame('BB', $b->read(100), 'A read past the end returns only what remains.'); + self::assertSame('', $b->read(100)); + self::assertTrue($b->eof()); + } + + public function testInPlaceAppendAccumulates() + { + $b = new TBufferStream(); + for ($i = 0; $i < 1000; $i++) { + $b->write('x'); + } + self::assertSame(1000, $b->getSize(), 'Repeated in-place appends accumulate correctly.'); + self::assertSame(str_repeat('x', 1000), (string) $b); + } + + public function testToStringIsNonDestructive() + { + $b = new TBufferStream(); + $b->write('keep'); + self::assertSame('keep', (string) $b); + self::assertSame('keep', (string) $b, 'Casting to string does not drain the buffer.'); + self::assertSame(4, $b->getSize()); + } + + public function testGetContentsDrainsAndIsCowSafe() + { + $b = new TBufferStream(); + $b->write('payload'); + $contents = $b->getContents(); + self::assertSame('payload', $contents); + self::assertSame(0, $b->getSize(), 'getContents() drains the buffer.'); + + // A subsequent write must not mutate the already-returned string (by-reference safety). + $b->write('other'); + self::assertSame('payload', $contents, 'The returned contents are independent of later writes.'); + } + + public function testReadNonPositiveLengthReturnsEmpty() + { + $b = new TBufferStream(); + $b->write('data'); + self::assertSame('', $b->read(0)); + self::assertSame('', $b->read(-5)); + self::assertSame(4, $b->getSize(), 'A non-positive read leaves the buffer intact.'); + } + + public function testCapabilitiesAndCloseIsTerminal() + { + $b = new TBufferStream(); + $b->write('data'); + self::assertTrue($b->isReadable()); + self::assertTrue($b->isWritable()); + self::assertFalse($b->isSeekable()); + + $b->close(); + self::assertNull($b->getSize(), 'A closed stream has an unknown size.'); + self::assertTrue($b->eof()); + self::assertFalse($b->isReadable(), 'A closed stream is no longer readable.'); + self::assertFalse($b->isWritable(), 'A closed stream is no longer writable.'); + } + + public function testPositionOperationsThrow() + { + $b = new TBufferStream(); + self::expectException(\RuntimeException::class); + $b->tell(); + } + + public function testSeekThrows() + { + $b = new TBufferStream(); + self::expectException(\RuntimeException::class); + $b->seek(0); + } + + public function testRewindThrows() + { + $b = new TBufferStream(); + self::expectException(\RuntimeException::class); + $b->rewind(); + } + + public function testCompactionPreservesDataAcrossTheChunkBoundary() + { + $b = new TBufferStream(); + $in = ''; + for ($i = 0; $i < 4000; $i++) { + $in .= sprintf('%09d|', $i); // 40000 unique bytes, so any mis-slice would show + } + $b->write($in); + + $out = ''; + while (!$b->eof()) { + $out .= $b->read(1337); // odd reads drive the pos >= CHUNK_SIZE compaction mid-drain + } + self::assertSame($in, $out, 'Compaction preserves every byte across the chunk boundary.'); + self::assertSame(0, $b->getSize()); + self::assertTrue($b->eof()); + } + + public function testInterleavedWriteBehindAConsumedPrefix() + { + $b = new TBufferStream(); + $b->write('abc'); + self::assertSame('ab', $b->read(2)); // consumes a prefix, leaving _bufferPos > 0 + $b->write('def'); // append behind the consumed prefix + self::assertSame(4, $b->getSize(), 'The unread count spans the old remainder and the new bytes.'); + self::assertSame('cdef', $b->read(4), 'FIFO order holds across a write behind a read.'); + self::assertTrue($b->eof()); + } + + public function testWriteMidLargeDrainStreamsTheTailIntact() + { + $b = new TBufferStream(); + $b->write(str_repeat('A', 20000)); + $b->read(9000); + $b->read(9000); // pos crosses CHUNK_SIZE and outweighs the remainder: compaction fires + $b->write('TAIL'); + + $rest = ''; + while (!$b->eof()) { + $rest .= $b->read(5000); + } + self::assertSame(2004, strlen($rest)); + self::assertSame('TAIL', substr($rest, -4), 'A write during a large drain comes out last, intact.'); + } + + public function testToStringAfterPartialReadReturnsTheUnreadRemainder() + { + $b = new TBufferStream(); + $b->write('abcdef'); + self::assertSame('ab', $b->read(2)); + self::assertSame('cdef', (string) $b); + self::assertSame('cdef', (string) $b, 'Casting after a partial read stays non-destructive.'); + self::assertSame(4, $b->getSize()); + self::assertSame('cd', $b->read(2)); + } + + public function testDetachReturnsNullAndIsTerminal() + { + $b = new TBufferStream(); + $b->write('data'); + self::assertNull($b->detach(), 'A buffer has no resource to return.'); + self::assertNull($b->getSize()); + self::assertTrue($b->eof()); + self::assertFalse($b->isReadable()); + self::assertFalse($b->isWritable()); + } + + public function testReadAfterCloseThrows() + { + $b = new TBufferStream(); + $b->write('data'); + $b->close(); + self::expectException(\RuntimeException::class); + $b->read(1); + } + + public function testWriteAfterCloseThrows() + { + $b = new TBufferStream(); + $b->close(); + self::expectException(\RuntimeException::class); + $b->write('x'); + } + + public function testGetContentsAfterDetachThrows() + { + $b = new TBufferStream(); + $b->write('data'); + $b->detach(); + self::expectException(\RuntimeException::class); + $b->getContents(); + } + + public function testEofFlipsBackToFalseAfterAWrite() + { + $b = new TBufferStream(); + self::assertTrue($b->eof()); + $b->write('x'); + self::assertFalse($b->eof()); + $b->read(1); + self::assertTrue($b->eof()); + $b->write('y'); + self::assertFalse($b->eof(), 'A write after draining reopens the buffer.'); + } + + public function testGetMetadataIsEmpty() + { + $b = new TBufferStream(); + self::assertSame([], $b->getMetadata()); + self::assertNull($b->getMetadata('uri')); + } + + public function testResetRevivesAClosedBufferAndIsFluent() + { + $b = new TBufferStream(); + $b->write('data'); + $b->close(); + self::assertFalse($b->isWritable(), 'The stream is closed.'); + + self::assertSame($b, $b->reset(), 'reset() returns the buffer for chaining.'); + self::assertTrue($b->isWritable(), 'reset() revives a closed buffer.'); + self::assertTrue($b->isReadable()); + self::assertSame(0, $b->getSize()); + self::assertTrue($b->eof()); + + $b->write('again'); + self::assertSame('again', $b->read(5), 'A revived buffer round-trips again.'); + } + + public function testResetDiscardsPendingBytesOnAnOpenBuffer() + { + $b = new TBufferStream(); + $b->write('unread'); + $b->reset(); + self::assertSame(0, $b->getSize(), 'reset() discards pending bytes.'); + self::assertTrue($b->eof()); + self::assertSame(2, $b->write('ok')); // still usable + } +}