Add decoding stream

src/DecodingInputStream.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace Kelunik\StreamingBase64;
+
+use Amp\ByteStream\InputStream;
+use Amp\Promise;
+use function Amp\call;
+
+class DecodingInputStream implements InputStream {
+    const TYPE_BASE64 = 0;
+    const TYPE_BASE64URL = 1;
+
+    private $source;
+    private $type;
+    private $buffer;
+
+    public function __construct(InputStream $source, int $type = self::TYPE_BASE64) {
+        $this->source = $source;
+        $this->type = $type;
+        $this->buffer = "";
+
+        if ($type < 0 || $type > 1) {
+            throw new \Error("Invalid type ({$type})");
+        }
+    }
+
+    /** @inheritdoc */
+    public function read(): Promise {
+        return call(function () {
+            if ($this->buffer === null) {
+                return null;
+            }
+
+            $data = yield $this->source->read();
+
+            if ($data === null) {
+                $buffer = $this->buffer;
+
+                $this->buffer = null;
+                $this->source = null;
+
+                if ($this->type === self::TYPE_BASE64) {
+                    return \base64_decode($buffer);
+                }
+
+                return \base64_decode(\strtr($buffer, "-_", "+/"));
+            }
+
+            $this->buffer .= $data;
+            $length = \strlen($this->buffer);
+            $buffer = \substr($this->buffer, 0, $length - $length % 4);
+            $this->buffer = \substr($this->buffer, $length - $length % 4);
+
+            if ($buffer !== "") {
+                if ($this->type === self::TYPE_BASE64) {
+                    return \base64_decode($buffer);
+                }
+
+                return \base64_decode(\strtr($buffer, "-_", "+/"));
+            }
+
+            return "";
+        });
+    }
+}

test/DecodingInputStreamTest.php

@@ -0,0 +1,44 @@
+<?php
+
+namespace Kelunik\StreamingBase64\Test;
+
+use Amp\ByteStream\InMemoryStream;
+use Amp\ByteStream\IteratorStream;
+use Amp\ByteStream\Message;
+use Amp\Loop;
+use Amp\PHPUnit\TestCase;
+use Kelunik\StreamingBase64\DecodingInputStream;
+use function Amp\Iterator\fromIterable;
+
+class DecodingInputStreamTest extends TestCase {
+    public function testInvalidType() {
+        $this->expectException(\Error::class);
+        new DecodingInputStream(new InMemoryStream(""), 2);
+    }
+
+    public function testBase64() {
+        Loop::run(function () {
+            $expected = "0d8de60626e522dc2c4bebb7e99b222b";
+            $array = ["M", "GQ4ZGU2MD", "YyNmU1MjJkYzJjNGJlYmI3ZTk5YjIyMmI", "="];
+            $stream = new IteratorStream(fromIterable($array, 10));
+
+            $decodingStream = new DecodingInputStream($stream, DecodingInputStream::TYPE_BASE64);
+            $result = yield new Message($decodingStream);
+
+            $this->assertSame($expected, $result);
+        });
+    }
+
+    public function testBase64Url() {
+        Loop::run(function () {
+            $expected = "0d8de60626e522dc2c4bebb7e99b222b";
+            $array = ["M", "GQ4ZGU2MD", "YyNmU1MjJkYzJjNGJlYmI3ZTk5YjIyMmI"];
+            $stream = new IteratorStream(fromIterable($array, 10));
+
+            $decodingStream = new DecodingInputStream($stream, DecodingInputStream::TYPE_BASE64URL);
+            $result = yield new Message($decodingStream);
+
+            $this->assertSame($expected, $result);
+        });
+    }
+}