IO Compressor - IStreamCodec - common interface for Streaming Codecs

framework/IO/Compression/IStreamCodec.php

@@ -0,0 +1,126 @@
+<?php
+
+/**
+ * IStreamCodec interface file.
+ *
+ * @author Brad Anderson <belisoful@icloud.com>
+ * @link https://github.com/pradosoft/prado
+ * @license https://github.com/pradosoft/prado/blob/master/LICENSE
+ */
+
+namespace Prado\IO\Compression;
+
+/**
+ * IStreamCodec interface.
+ *
+ * An incremental byte-stream codec context, modeled on PHP's own {@see deflate_init()}/
+ * {@see deflate_add()} (and {@see inflate_init()}/{@see inflate_add()}): the implementing
+ * class hands back a fresh context, {@see add()} pushes each input chunk and returns
+ * whatever output is ready, and {@see finish()} flushes the trailing state.  A context is
+ * single-use and single-direction, so encoding and decoding are separate contexts.
+ *
+ * A context holds only the bounded state its algorithm needs — a carry buffer, a
+ * dictionary, a partial byte — so it transforms a stream of any size in constant memory.
+ * Chunk boundaries are invisible to the result: the same bytes fed as one call or as many
+ * produce the same output.
+ *
+ * The interface exists so one codec implementation serves both consumers in the IO layer:
+ *
+ * | Consumer | How it drives the codec |
+ * |----------|-------------------------|
+ * | {@see ICompressor} (whole string) | `add($all) . finish()` |
+ * | {@see \Prado\IO\Filter\TStreamCodecFilter} (streaming) | `process()` calls {@see add()}, `finish()` calls {@see finish()} |
+ *
+ * A codec is a plain class holding its own state:
+ *
+ * ```php
+ * class TUpperPairCodec implements IStreamCodec
+ * {
+ *     private string $_carry = '';
+ *
+ *     public function add(string $data): string
+ *     {
+ *         $buffer = $this->_carry . $data;
+ *         $whole = intdiv(strlen($buffer), 2) * 2;   // emit only complete pairs
+ *         $this->_carry = substr($buffer, $whole);   // hold the odd byte for the next chunk
+ *         return strtoupper(substr($buffer, 0, $whole));
+ *     }
+ *
+ *     public function finish(): string
+ *     {
+ *         $tail = $this->_carry;
+ *         $this->_carry = '';
+ *         return strtoupper($tail);
+ *     }
+ * }
+ *
+ * $codec = new TUpperPairCodec();
+ * $out  = $codec->add('ab');
+ * $out .= $codec->add('c');     // 'c' is held; nothing emitted yet
+ * $out .= $codec->finish();     // 'ABC'
+ * ```
+ *
+ * The same context drives a stream filter, so the streaming path duplicates no logic:
+ *
+ * ```php
+ * class TUpperPairFilter extends TStreamCodecFilter
+ * {
+ *     private IStreamCodec $_codec;
+ *
+ *     public static function getFilterName(): string
+ *     {
+ *         return 'prado.upperpair';
+ *     }
+ *
+ *     public function onCreate(): bool
+ *     {
+ *         $this->_codec = new TUpperPairCodec();
+ *         return true;
+ *     }
+ *
+ *     protected function process(string $data): string
+ *     {
+ *         return $this->_codec->add($data);
+ *     }
+ *
+ *     protected function finish(): string
+ *     {
+ *         return $this->_codec->finish();
+ *     }
+ * }
+ *
+ * TUpperPairFilter::registerOnce();
+ * $stream = TStream::fromString('abc');
+ * $stream->appendFilter(TUpperPairFilter::getFilterName(), STREAM_FILTER_READ);
+ * echo $stream->getContents();   // 'ABC'
+ * ```
+ *
+ * @author Brad Anderson <belisoful@icloud.com>
+ * @since 4.4.0
+ */
+interface IStreamCodec
+{
+	/**
+	 * Pushes a chunk of input and returns the output produced so far.  Bounded state is
+	 * carried to the next call, so a field split across chunks is handled correctly and an
+	 * empty chunk is a no-op.
+	 *
+	 * ```php
+	 * $codec->add('ab') . $codec->add('c');   // same bytes as $codec->add('abc')
+	 * ```
+	 * @param string $data The input chunk (may be '').
+	 * @return string The output produced from this chunk (may be '').
+	 */
+	public function add(string $data): string;
+
+	/**
+	 * Flushes any pending state and returns the final output.  After this the context is
+	 * spent; further {@see add()} calls are not defined.
+	 *
+	 * ```php
+	 * $encoded = $codec->add($data) . $codec->finish();   // the whole-string form
+	 * ```
+	 * @return string The final output (may be '').
+	 */
+	public function finish(): string;
+}

framework/classes.php

@@ -269,6 +269,7 @@
 'TPhpStreamBehavior' => 'Prado\IO\Behaviors\TPhpStreamBehavior',
 'TStreamNoSeekBehavior' => 'Prado\IO\Behaviors\TStreamNoSeekBehavior',
 'ICompressor' => 'Prado\IO\Compression\ICompressor',
+'IStreamCodec' => 'Prado\IO\Compression\IStreamCodec',
 'TBrotliCompressor' => 'Prado\IO\Compression\TBrotliCompressor',
 'TBuiltinCompressor' => 'Prado\IO\Compression\TBuiltinCompressor',
 'TBzip2Compressor' => 'Prado\IO\Compression\TBzip2Compressor',

tests/unit/IO/Compression/IStreamCodecTest.php

@@ -0,0 +1,182 @@
+<?php
+
+use Prado\IO\Compression\ICompressor;
+use Prado\IO\Compression\IStreamCodec;
+use Prado\IO\Filter\TStreamCodecFilter;
+use Prado\IO\TStream;
+
+/**
+ * A codec with real carry state: it emits complete byte pairs uppercased and holds an odd
+ * byte until the next chunk or {@see finish()}.  An implementation that ignored the carry
+ * would produce different bytes for different chunkings, so the contract is testable.
+ */
+class UpperPairCodec implements IStreamCodec
+{
+	private string $_carry = '';
+
+	public function add(string $data): string
+	{
+		$buffer = $this->_carry . $data;
+		$whole = intdiv(strlen($buffer), 2) * 2;
+		$this->_carry = substr($buffer, $whole);
+		return strtoupper(substr($buffer, 0, $whole));
+	}
+
+	public function finish(): string
+	{
+		$tail = $this->_carry;
+		$this->_carry = '';
+		return strtoupper($tail);
+	}
+}
+
+/** The whole-string consumer: ICompressor over one codec context per call. */
+class UpperPairCompressor implements ICompressor
+{
+	public static function compress(string $data): string
+	{
+		$codec = new UpperPairCodec();
+		return $codec->add($data) . $codec->finish();
+	}
+
+	public static function decompress(string $data): string
+	{
+		return strtolower($data);
+	}
+}
+
+/** The streaming consumer: a stream filter driving the same codec. */
+class UpperPairFilter extends TStreamCodecFilter
+{
+	private IStreamCodec $_codec;
+
+	public static function getFilterName(): string
+	{
+		return 'prado.test.upperpair';
+	}
+
+	public function onCreate(): bool
+	{
+		$this->_codec = new UpperPairCodec();
+		return true;
+	}
+
+	protected function process(string $data): string
+	{
+		return $this->_codec->add($data);
+	}
+
+	protected function finish(): string
+	{
+		return $this->_codec->finish();
+	}
+}
+
+class IStreamCodecTest extends PHPUnit\Framework\TestCase
+{
+	public function testTheContractShape()
+	{
+		$codec = new UpperPairCodec();
+		self::assertInstanceOf(IStreamCodec::class, $codec);
+		$ref = new \ReflectionClass(IStreamCodec::class);
+		self::assertSame(['add', 'finish'], array_map(fn ($m) => $m->getName(), $ref->getMethods()), 'The context is add() plus finish().');
+	}
+
+	public function testAddThenFinishProducesTheWholeOutput()
+	{
+		$codec = new UpperPairCodec();
+		self::assertSame('AB', $codec->add('abc'), 'Only the complete pair is emitted.');
+		self::assertSame('C', $codec->finish(), 'finish() flushes the held byte.');
+	}
+
+	public function testChunkBoundariesAreInvisibleToTheResult()
+	{
+		$data = 'the quick brown fox jumps over the lazy dog';
+		$whole = (new UpperPairCodec())->add($data);
+		$whole .= '';   // the reference output below is assembled the same way
+		$reference = strtoupper($data);
+
+		foreach ([[1], [2], [3], [5], [8], [1000]] as [$size]) {
+			$codec = new UpperPairCodec();
+			$out = '';
+			foreach (str_split($data, $size) as $chunk) {
+				$out .= $codec->add($chunk);
+			}
+			$out .= $codec->finish();
+			self::assertSame($reference, $out, "chunk size {$size} yields the same bytes");
+		}
+		self::assertNotSame('', $whole);
+	}
+
+	public function testEmptyChunksAreNoOps()
+	{
+		$codec = new UpperPairCodec();
+		self::assertSame('', $codec->add(''));
+		self::assertSame('AB', $codec->add('ab'));
+		self::assertSame('', $codec->add(''), 'An empty chunk emits nothing and disturbs no state.');
+		self::assertSame('', $codec->finish(), 'Nothing is pending after a whole pair.');
+	}
+
+	public function testEmptyInputFinishesEmpty()
+	{
+		$codec = new UpperPairCodec();
+		self::assertSame('', $codec->add(''));
+		self::assertSame('', $codec->finish());
+	}
+
+	public function testContextsAreIndependent()
+	{
+		$a = new UpperPairCodec();
+		$b = new UpperPairCodec();
+		$a->add('x');                                   // 'x' held in $a only
+		self::assertSame('YZ', $b->add('yz'), 'A second context carries none of the first context state.');
+		self::assertSame('', $b->finish());
+		self::assertSame('X', $a->finish());
+	}
+
+	// ---- the two consumers share the one implementation ------------------------
+
+	public function testWholeStringConsumerMatchesTheCodec()
+	{
+		$data = 'abcde';
+		$codec = new UpperPairCodec();
+		self::assertSame($codec->add($data) . $codec->finish(), UpperPairCompressor::compress($data));
+		self::assertSame('ABCDE', UpperPairCompressor::compress($data));
+	}
+
+	public function testStreamFilterConsumerMatchesTheCodec()
+	{
+		UpperPairFilter::registerOnce();
+		self::assertTrue(UpperPairFilter::isRegistered(UpperPairFilter::getFilterName()));
+
+		$data = 'abcde';   // odd length, so the filter's finish() must flush the carry
+		$stream = TStream::fromString($data);
+		$stream->appendFilter(UpperPairFilter::getFilterName(), STREAM_FILTER_READ);
+		self::assertSame(UpperPairCompressor::compress($data), $stream->getContents(), 'The filter and the whole-string form agree.');
+		$stream->close();
+	}
+
+	public function testStreamFilterAgreesAcrossReadSizes()
+	{
+		UpperPairFilter::registerOnce();
+		$data = 'the quick brown fox';
+		$expected = strtoupper($data);
+
+		foreach ([1, 3, 7, 8192] as $readSize) {
+			$handle = fopen('php://temp', 'r+b');
+			fwrite($handle, $data);
+			rewind($handle);
+			stream_filter_append($handle, UpperPairFilter::getFilterName(), STREAM_FILTER_READ);
+			$out = '';
+			while (!feof($handle)) {
+				$piece = fread($handle, $readSize);
+				if ($piece === false) {
+					break;
+				}
+				$out .= $piece;
+			}
+			fclose($handle);
+			self::assertSame($expected, $out, "read size {$readSize} yields the same bytes");
+		}
+	}
+}