Add LineReader (#64)

lib/LineReader.php

@@ -0,0 +1,62 @@
+<?php
+
+namespace Amp\ByteStream;
+
+use Amp\Promise;
+use function Amp\call;
+
+final class LineReader
+{
+    /** @var string */
+    private $buffer = "";
+
+    /** @var InputStream */
+    private $source;
+
+    public function __construct(InputStream $inputStream)
+    {
+        $this->source = $inputStream;
+    }
+
+    /**
+     * @return Promise<string|null>
+     */
+    public function readLine(): Promise
+    {
+        return call(function () {
+            if (($pos = \strpos($this->buffer, "\n")) !== false) {
+                $line = \substr($this->buffer, 0, $pos);
+                $this->buffer = \substr($this->buffer, $pos + 1);
+                return \rtrim($line, "\r");
+            }
+
+            while (null !== $chunk = yield $this->source->read()) {
+                $this->buffer .= $chunk;
+
+                if (($pos = \strpos($this->buffer, "\n")) !== false) {
+                    $line = \substr($this->buffer, 0, $pos);
+                    $this->buffer = \substr($this->buffer, $pos + 1);
+                    return \rtrim($line, "\r");
+                }
+            }
+
+            if ($this->buffer === "") {
+                return null;
+            }
+
+            $line = $this->buffer;
+            $this->buffer = "";
+            return \rtrim($line, "\r");
+        });
+    }
+
+    public function getBuffer(): string
+    {
+        return $this->buffer;
+    }
+
+    public function clearBuffer()
+    {
+        $this->buffer = "";
+    }
+}

test/LineReaderTest.php

@@ -0,0 +1,94 @@
+<?php /** @noinspection PhpUnhandledExceptionInspection */
+
+namespace Amp\ByteStream;
+
+use Amp\Iterator;
+use Amp\PHPUnit\TestCase;
+use function Amp\call;
+use function Amp\Promise\wait;
+
+class LineReaderTest extends TestCase
+{
+    public function testSingleLine()
+    {
+        $this->check(["abc"], ["abc"]);
+    }
+
+    public function testMultiLineLf()
+    {
+        $this->check(["abc\nef"], ["abc", "ef"]);
+    }
+
+    public function testMultiLineCrLf()
+    {
+        $this->check(["abc\r\nef"], ["abc", "ef"]);
+    }
+
+    public function testMultiLineEmptyNewlineStart()
+    {
+        $this->check(["\r\nabc\r\nef\r\n"], ["", "abc", "ef"]);
+    }
+
+    public function testMultiLineEmptyNewlineEnd()
+    {
+        $this->check(["abc\r\nef\r\n"], ["abc", "ef"]);
+    }
+
+    public function testMultiLineEmptyNewlineMiddle()
+    {
+        $this->check(["abc\r\n\r\nef\r\n"], ["abc", "", "ef"]);
+    }
+
+    public function testEmpty()
+    {
+        $this->check([], []);
+    }
+
+    public function testEmptyCrLf()
+    {
+        $this->check(["\r\n"], [""]);
+    }
+
+    public function testEmptyCr()
+    {
+        $this->check(["\r"], [""]);
+    }
+
+    public function testMultiLineSlow()
+    {
+        $this->check(["a", "bc", "\r", "\n\r\nef\r", "\n"], ["abc", "", "ef"]);
+    }
+
+    public function testClearBuffer()
+    {
+        wait(call(static function () {
+            $inputStream = new IteratorStream(Iterator\fromIterable(["a\nb\nc"]));
+
+            $reader = new LineReader($inputStream);
+            self::assertSame("a", yield $reader->readLine());
+            self::assertSame("b\nc", $reader->getBuffer());
+
+            $reader->clearBuffer();
+
+            self::assertSame("", $reader->getBuffer());
+            self::assertNull(yield $reader->readLine());
+        }));
+    }
+
+    private function check(array $chunks, array $expectedLines)
+    {
+        wait(call(static function () use ($chunks, $expectedLines) {
+            $inputStream = new IteratorStream(Iterator\fromIterable($chunks));
+
+            $reader = new LineReader($inputStream);
+            $lines = [];
+
+            while (null !== $line = yield $reader->readLine()) {
+                $lines[] = $line;
+            }
+
+            self::assertSame($expectedLines, $lines);
+            self::assertSame("", $reader->getBuffer());
+        }));
+    }
+}