Merge pull request #359 from thekid/feature/filter-streams

Integrate stream filters into io.streams

Author: thekid

src/main/php/io/streams/FilterInputStream.class.php

@@ -0,0 +1,44 @@
+<?php namespace io\streams;
+
+/**
+ * Input stream that runs input through a given filter
+ *
+ * @test  io.unittest.FilterInputStreamTest
+ */
+class FilterInputStream extends FilterStream implements InputStream {
+  const MODE= STREAM_FILTER_READ;
+
+  static function __static() { }
+
+  /**
+   * Creates a new instance
+   * 
+   * @param io.streams.InputStream $in
+   * @param string|callable|string[]|callable[]|[:string] $filters
+   */
+  public function __construct(InputStream $in, $filters) {
+    $this->fd= Streams::readableFd($in);
+    foreach (is_array($filters) ? $filters : [$filters] as $key => $filter) {
+      is_array($filter) ? $this->append($key, $filter) : $this->append($filter);
+    }
+  }
+
+  /**
+   * Read bytes
+   *
+   * @param  int $limit default 8192
+   * @return string
+   */
+  public function read($limit= 8192) {
+    return fread($this->fd, $limit);
+  }
+
+  /**
+   * Availably bytes
+   *
+   * @return int
+   */
+  public function available() {
+    return feof($this->fd) ? 0 : 1;
+  }
+}

src/main/php/io/streams/FilterOutputStream.class.php

@@ -0,0 +1,44 @@
+<?php namespace io\streams;
+
+/**
+ * Output stream that runs input through a given filter
+ *
+ * @test  io.unittest.FilterOutputStreamTest
+ */
+class FilterOutputStream extends FilterStream implements OutputStream {
+  const MODE= STREAM_FILTER_WRITE;
+
+  static function __static() { }
+
+  /**
+   * Creates a new instance
+   * 
+   * @param io.streams.OutputStream $out
+   * @param string|callable|string[]|callable[]|[:string] $filters
+   */
+  public function __construct(OutputStream $out, $filters= []) {
+    $this->fd= Streams::writeableFd($out);
+    foreach (is_array($filters) ? $filters : [$filters] as $key => $filter) {
+      is_array($filter) ? $this->append($key, $filter) : $this->append($filter);
+    }
+  }
+
+  /**
+   * Write bytes
+   *
+   * @param  string $bytes
+   * @return void
+   */
+  public function write($bytes) {
+    fwrite($this->fd, $bytes);
+  }
+
+  /**
+   * Flush
+   *
+   * @return void
+   */
+  public function flush() {
+    fflush($this->fd);
+  }
+}

src/main/php/io/streams/FilterStream.class.php

@@ -0,0 +1,86 @@
+<?php namespace io\streams;
+
+use ReflectionFunction, php_user_filter;
+use lang\IllegalArgumentException;
+
+/** @see https://www.php.net/manual/en/filters.php */
+abstract class FilterStream {
+  public static $filters= [];
+  private static $id= 0;
+  protected $fd;
+  private $remove= [];
+
+  static function __static() {
+
+    // Suppress "Declaration should be compatible" in PHP 7.4
+    stream_filter_register('iostrl.*', get_class(@new class() extends php_user_filter {
+      public function filter($in, $out, &$consumed, bool $closing): int {
+        while ($bucket= stream_bucket_make_writeable($in)) {
+          $consumed+= $bucket->datalen;
+          $bucket->data= FilterStream::$filters[$this->filtername]($bucket->data);
+          null === $bucket->data || stream_bucket_append($out, $bucket);
+        }
+        return PSFS_PASS_ON;
+      }
+    }));
+    stream_filter_register('iostrf.*', get_class(@new class() extends php_user_filter {
+      public function filter($in, $out, &$consumed, bool $closing): int {
+        return FilterStream::$filters[$this->filtername]($in, $out, $consumed, $closing);
+      }
+    }));
+  }
+
+  /**
+   * Appends a filter and returns its handle
+   *
+   * @param  string|callable $filter
+   * @param  array $parameters
+   * @return mixed
+   * @throws lang.IllegalArgumentException
+   */
+  public function append($filter, $parameters= []) {
+    if (is_string($filter)) {
+      $name= $filter;
+    } else {
+      $f= new ReflectionFunction($filter);
+      $this->remove[]= $name= (1 === $f->getNumberOfParameters() ? 'iostrl.' : 'iostrf.').(++self::$id);
+      self::$filters[$name]= $filter;
+    }
+
+    if (!($handle= stream_filter_append($this->fd, $name, static::MODE, $parameters))) {
+      throw new IllegalArgumentException('Could not append stream filter '.$name);
+    }
+    return $handle;
+  }
+
+  /**
+   * Removes a stream filter
+   *
+   * @param  mixed $handle
+   * @return void
+   * @throws lang.IllegalArgumentException
+   */
+  public function remove($handle) {
+    if (!stream_filter_remove($handle)) {
+      throw new IllegalArgumentException('Could not remove stream filter '.$name);
+    }
+  }
+
+  /**
+   * Close stream
+   *
+   * @return void
+   */
+  public function close() {
+    fclose($this->fd);
+    $this->fd= null;
+  }
+
+  /** Ensures stream is closed */
+  public function __destruct() {
+    foreach ($this->remove as $filter) {
+      unset(self::$filters[$filter]);
+    }
+    $this->fd && $this->close();
+  }
+}

src/test/php/io/unittest/FilterInputStreamTest.class.php

@@ -0,0 +1,50 @@
+<?php namespace io\unittest;
+
+use io\streams\{FilterInputStream, MemoryInputStream, Streams};
+use lang\IllegalStateException;
+use test\{Assert, Expect, Test, Values};
+
+class FilterInputStreamTest {
+
+  /** Test helper */
+  private function read($input, $filter) {
+    return Streams::readAll(new FilterInputStream(new MemoryInputStream($input), $filter));
+  }
+
+  #[Test, Values([['str_rot13', 'string.rot13'], ['base64_encode', 'convert.base64-decode'], ['quoted_printable_encode', 'convert.quoted-printable-decode']])]
+  public function builtin($encode, $filter) {
+    Assert::equals('Test', $this->read($encode('Test'), $filter));
+  }
+
+  #[Test]
+  public function chained() {
+    Assert::equals('Test', $this->read(base64_encode(str_rot13('Test')), [
+      'convert.base64-decode',
+      'string.rot13',
+    ]));
+  }
+
+  #[Test]
+  public function lambda() {
+    Assert::equals('test', $this->read('TEST', fn($chunk) => strtolower($chunk)));
+  }
+
+  #[Test, Expect(IllegalStateException::class)]
+  public function lambda_raising_error() {
+    $this->read('TEST', function($chunk) {
+      throw new IllegalStateException('Test');
+    });
+  }
+
+  #[Test]
+  public function user_filter() {
+    Assert::equals('test', $this->read('TEST', function($in, $out, &$consumed, $closing) {
+      while ($bucket= stream_bucket_make_writeable($in)) {
+        $consumed+= $bucket->datalen;
+        $bucket->data= strtolower($bucket->data);
+        stream_bucket_append($out, $bucket);
+      }
+      return PSFS_PASS_ON;
+    }));
+  }
+}

src/test/php/io/unittest/FilterOutputStreamTest.class.php

@@ -0,0 +1,85 @@
+<?php namespace io\unittest;
+
+use io\streams\{FilterOutputStream, MemoryOutputStream};
+use test\verify\Runtime;
+use test\{Assert, Test, Values};
+
+class FilterOutputStreamTest {
+
+  /** Test helper */
+  private function write($input, $filter) {
+    $out= new MemoryOutputStream();
+
+    $fixture= new FilterOutputStream($out, $filter);
+    $fixture->write($input);
+    $fixture->close();
+
+    return $out->bytes();
+  }
+
+  #[Test, Values([['str_rot13', 'string.rot13'], ['base64_encode', 'convert.base64-encode'], ['quoted_printable_encode', 'convert.quoted-printable-encode']])]
+  public function builtin($encode, $filter) {
+    Assert::equals($encode('Test'), $this->write('Test', $filter));
+  }
+
+  #[Test]
+  public function chained() {
+    Assert::equals(base64_encode(str_rot13('Test')), $this->write('Test', [
+      'string.rot13',
+      'convert.base64-encode',
+    ]));
+  }
+
+  #[Test]
+  public function lambda() {
+    Assert::equals('test', $this->write('TEST', fn($chunk) => strtolower($chunk)));
+  }
+
+  #[Test]
+  public function user_filter() {
+    Assert::equals('test', $this->write('TEST', function($in, $out, &$consumed, $closing) {
+      while ($bucket= stream_bucket_make_writeable($in)) {
+        $consumed+= $bucket->datalen;
+        $bucket->data= strtolower($bucket->data);
+        stream_bucket_append($out, $bucket);
+      }
+      return PSFS_PASS_ON;
+    }));
+  }
+
+  #[Test]
+  public function append() {
+    $out= new MemoryOutputStream();
+    $fixture= new FilterOutputStream($out);
+
+    $toupper= $fixture->append('string.toupper');
+    $rot13= $fixture->append('string.rot13');
+    $fixture->write('test');
+
+    $fixture->remove($rot13);
+    $fixture->write('e');
+
+    $fixture->remove($toupper);
+    $fixture->write('d');
+
+    $fixture->close();
+
+    Assert::equals(str_rot13(strtoupper('test')).strtoupper('e').'d', $out->bytes());
+  }
+
+  #[Test]
+  public function base64_line_length() {
+    Assert::equals(
+      "VGhpcyBp\r\ncyBhIHRl\r\nc3QuCg==",
+      $this->write("This is a test.\n", ['convert.base64-encode' => ['line-length' => 8]])
+    );
+  }
+
+  #[Test, Runtime(extensions: ['zlib'])]
+  public function deflate_level() {
+    Assert::equals(
+      gzdeflate('This is a test. This also.', 6),
+      $this->write('This is a test. This also.', ['zlib.deflate' => ['level' => 6]])
+    );
+  }
+}