Add a Blob class as a bridge between string, bytes and streams

src/main/php/io/Blob.class.php

@@ -0,0 +1,148 @@
+<?php namespace io;
+
+use IteratorAggregate, Traversable;
+use io\streams\{InputStream, IterableInputStream, Streams};
+use lang\{Value, IllegalArgumentException};
+use util\{Bytes, Objects};
+
+/** @test io.unittest.BlobTest */
+class Blob implements IteratorAggregate, Value {
+  private $parts;
+  private $iterator= null;
+  public $meta= [];
+
+  /**
+   * Creates a new blob from parts
+   *
+   * @param  iterable|string|util.Bytes|io.streams.InputStream $parts
+   * @param  [:var] $meta
+   * @throws lang.IllegalArgumentException
+   */
+  public function __construct($parts= [], array $meta= []) {
+    if ($parts instanceof InputStream) {
+      $this->iterator= function() {
+        static $started= false;
+
+        return (function() use(&$started) {
+          $started ? Streams::seek($this->parts, 0) : $started= true;
+          while ($this->parts->available()) {
+            yield $this->parts->read();
+          }
+        })();
+      };
+    } else if ($parts instanceof Bytes || is_string($parts)) {
+      $this->iterator= fn() => (function() { yield (string)$this->parts; })();
+    } else if (is_iterable($parts)) {
+      $this->iterator= fn() => (function() {
+        foreach ($this->parts as $part) {
+          yield (string)$part;
+        }
+      })();
+    } else {
+      throw new IllegalArgumentException(sprintf(
+        'Expected iterable|string|util.Bytes|io.streams.InputStream, have %s',
+        typeof($parts)
+      ));
+    }
+
+    $this->parts= $parts;
+    $this->meta= $meta;
+  }
+
+  /** @return iterable */
+  public function getIterator(): Traversable { return ($this->iterator)(); }
+
+  /** @return util.Bytes */
+  public function bytes() { 
+    return $this->parts instanceof Bytes
+      ? $this->parts
+      : new Bytes(...($this->iterator)())
+    ;
+  }
+
+  /** @return io.streams.InputStream */
+  public function stream() {
+    return $this->parts instanceof InputStream
+      ? $this->parts
+      : new IterableInputStream(($this->iterator)())
+    ;
+  }
+
+  /** Creates a new blob with the given filter applied */
+  public function encoded(string $filter): self {
+    $it= function() use($filter) {
+      $fd= Streams::readableFd($this->stream());
+      if (!stream_filter_append($fd, $filter, STREAM_FILTER_READ)) {
+        fclose($fd);
+        throw new OperationNotSupportedException('Cannot stream '.$filter);
+      }
+
+      try {
+        do {
+          yield fread($fd, 8192);
+        } while (!feof($fd));
+      } finally {
+        fclose($fd);
+      }
+    };
+
+    $meta= $this->meta;
+    $meta['encoding']??= [];
+    $meta['encoding'][]= $filter;
+
+    return new self($it(), $meta);
+  }
+
+  /** @return iterable */
+  public function slices(int $size= 8192) {
+    $it= ($this->iterator)();
+    $it->rewind();
+    while ($it->valid()) {
+      $slice= $it->current();
+      $length= strlen($slice);
+      $offset= 0;
+
+      while ($length < $size) {
+        $it->next();
+        $slice.= $it->current();
+        if (!$it->valid()) break;
+      }
+
+      while ($length - $offset > $size) {
+        yield substr($slice, $offset, $size);
+        $offset+= $size;
+      }
+
+      yield $offset ? substr($slice, $offset) : $slice;
+      $it->next();
+    }
+  }
+
+  /** @return string */
+  public function hashCode() { return 'B'.Objects::hashOf($this->parts); }
+
+  /** @return string */
+  public function toString() { return nameof($this).'('.Objects::stringOf($this->parts).')'; }
+
+  /**
+   * Comparison
+   *
+   * @param  var $value
+   * @return int
+   */
+  public function compareTo($value) {
+    return $value instanceof self
+      ? Objects::compare($this->parts, $value->parts)
+      : 1
+    ;
+  }
+
+  /** @return string */
+  public function __toString() {
+    $bytes= '';
+    foreach (($this->iterator)() as $chunk) {
+      $bytes.= $chunk;
+    }
+    return $bytes;
+  }
+}

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

@@ -0,0 +1,63 @@
+<?php namespace io\streams;
+
+use Iterator, Closure, Traversable;
+use lang\IllegalArgumentException;
+
+/** @test io.unittest.IterableInputStreamTest */
+class IterableInputStream implements InputStream {
+  private $iterator;
+  private $buffer= null;
+
+  /** @param iterable|function(): Iterator $input */
+  public function __construct($input) {
+    if ($input instanceof Iterator) {
+      $this->iterator= $input;
+    } else if ($input instanceof Closure) {
+      $this->iterator= cast($input(), Iterator::class);
+    } else if (is_iterable($input)) {
+      $this->iterator= (function() use($input) { yield from $input; })();
+    } else {
+      throw new IllegalArgumentException('Expected iterable|function(): Iterator, have '.typeof($input));
+    }
+    $this->iterator->rewind();
+  }
+
+  /** @return int */
+  public function available() {
+    if (null !== $this->buffer) {
+      return strlen($this->buffer);
+    } else if ($this->iterator->valid()) {
+      $this->buffer= $this->iterator->current();
+      $this->iterator->next();
+      return strlen($this->buffer);
+    } else {
+      return 0;
+    }
+  }
+
+  /**
+   * Reads up to a given limit
+   *
+   * @param  int $limit
+   * @return string
+   */
+  public function read($limit= 8192) {
+    if (null !== $this->buffer) {
+      // Continue draining the buffer
+    } else if ($this->iterator->valid()) {
+      $this->buffer= $this->iterator->current();
+      $this->iterator->next();
+    } else {
+      return '';
+    }
+
+    $chunk= substr($this->buffer, 0, $limit);
+    $this->buffer= $limit >= strlen($this->buffer) ? null : substr($this->buffer, $limit);
+    return $chunk;
+  }
+
+  /** @return void */
+  public function close() {
+    // NOOP
+  }
+}

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

@@ -1,7 +1,6 @@
 <?php namespace io\streams;
 
-use io\FileNotFoundException;
-use io\IOException;
+use io\{FileNotFoundException, OperationNotSupportedException, IOException};
 
 /**
  * Wraps I/O streams into PHP streams
@@ -134,6 +133,23 @@ public static function readAll(InputStream $s) {
     return $r;
   }
 
+  /**
+   * Read an IOElements' contents completely into a buffer in a single call.
+   *
+   * @param   io.streams.InputStream $s
+   * @param   int $offset
+   * @param   int $whence default SEEK_SET (one of SEEK_[SET|CUR|END])
+   * @return  void
+   * @throws  io.IOException
+   */
+  public static function seek($s, $offset, $whence= SEEK_SET) {
+    if ($s instanceof Seekable) {
+      $s->seek($offset, $whence);
+    } else {
+      throw new OperationNotSupportedException('Cannot seek instances of '.nameof($s));
+    }
+  }
+
   /**
    * Callback for fopen
    *

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

@@ -0,0 +1,111 @@
+<?php namespace io\unittest;
+
+use ArrayObject;
+use io\streams\{MemoryInputStream, InputStream};
+use io\{Blob, OperationNotSupportedException};
+use lang\IllegalArgumentException;
+use test\verify\Runtime;
+use test\{Assert, Expect, Test, Values};
+use util\Bytes;
+
+class BlobTest {
+
+  /** @return iterable */
+  private function cases() {
+    yield [new Blob(), []];
+    yield [new Blob('Test'), ['Test']];
+    yield [new Blob(['Über']), ['Über']];
+    yield [new Blob([new Blob(['Test']), 'ed']), ['Test', 'ed']];
+    yield [new Blob(['Test', 'ed']), ['Test', 'ed']];
+    yield [new Blob((function() { yield 'Test'; yield 'ed'; })()), ['Test', 'ed']];
+    yield [new Blob(new ArrayObject(['Test', 'ed'])), ['Test', 'ed']];
+    yield [new Blob(new Bytes('Test')), ['Test']];
+    yield [new Blob(new MemoryInputStream('Test')), ['Test']];
+  }
+
+  #[Test]
+  public function can_create() {
+    new Blob();
+  }
+
+  #[Test, Expect(IllegalArgumentException::class)]
+  public function not_from_null() {
+    new Blob(null);
+  }
+
+  #[Test]
+  public function meta_empty_by_default() {
+    Assert::equals([], (new Blob('Test'))->meta);
+  }
+
+  #[Test]
+  public function meta() {
+    $meta= ['type' => 'text/plain'];
+    Assert::equals($meta, (new Blob('Test', $meta))->meta);
+  }
+
+  #[Test, Values(from: 'cases')]
+  public function iteration($fixture, $expected) {
+    Assert::equals($expected, iterator_to_array($fixture));
+  }
+
+  #[Test, Values(from: 'cases')]
+  public function bytes($fixture, $expected) {
+    Assert::equals(new Bytes($expected), $fixture->bytes());
+  }
+
+  #[Test, Values(from: 'cases')]
+  public function stream($fixture, $expected) {
+    $stream= $fixture->stream();
+    $data= [];
+    while ($stream->available()) {
+      $data[]= $stream->read();
+    }
+    Assert::equals($expected, $data);
+  }
+
+  #[Test, Values(from: 'cases')]
+  public function string_cast($fixture, $expected) {
+    Assert::equals(implode('', $expected), (string)$fixture);
+  }
+
+  #[Test, Values([[1, ['T', 'e', 's', 't']], [2, ['Te', 'st']], [3, ['Tes', 't']], [4, ['Test']]])]
+  public function slices($size, $expected) {
+    Assert::equals($expected, iterator_to_array((new Blob('Test'))->slices($size)));
+  }
+
+  #[Test]
+  public function fill_slice() {
+    Assert::equals(['Test'], iterator_to_array((new Blob(['Te', 'st']))->slices()));
+  }
+
+  #[Test]
+  public function fetch_slice_twice() {
+    $fixture= new Blob('Test');
+
+    Assert::equals(['Test'], iterator_to_array($fixture->slices()));
+    Assert::equals(['Test'], iterator_to_array($fixture->slices()));
+  }
+
+  #[Test]
+  public function cannot_fetch_slices_twice_from_non_seekable() {
+    $fixture= new Blob(new class() implements InputStream {
+      private $input= ['Test'];
+      public function available() { return strlen(current($this->input)); }
+      public function read($limit= 8192) { return array_shift($this->input); }
+      public function close() { $this->input= []; }
+    });
+    iterator_to_array($fixture->slices());
+
+    Assert::throws(OperationNotSupportedException::class, fn() => iterator_to_array($fixture->slices()));
+  }
+
+  /** @see https://bugs.php.net/bug.php?id=77069 */
+  #[Test, Runtime(php: '>=7.4.14')]
+  public function base64_encoded() {
+    $base64= (new Blob('Test'))->encoded('convert.base64-encode');
+
+    Assert::equals(['convert.base64-encode'], $base64->meta['encoding']);
+    Assert::equals('VGVzdA==', (string)$base64);
+  }
+}