Merge

lib/LineReader.php

@@ -54,4 +54,9 @@ final class LineReader
     {
         return $this->buffer;
     }
+
+    public function clearBuffer()
+    {
+        $this->buffer = "";
+    }
 }

lib/functions.php

@@ -2,7 +2,9 @@
 
 namespace Amp\ByteStream;
 
+use Amp\Iterator;
 use Amp\Loop;
+use Amp\Producer;
 use Amp\Promise;
 use function Amp\call;
 
@@ -44,7 +46,7 @@ function pipe(InputStream $source, OutputStream $destination): Promise
 }
 
 /**
- * @param \Amp\ByteStream\InputStream  $source
+ * @param \Amp\ByteStream\InputStream $source
  *
  * @return \Amp\Promise
  */
@@ -99,6 +101,7 @@ function getOutputBufferStream(): ResourceOutputStream
 
     return $stream;
 }
+
 /**
  * The STDIN stream for the process associated with the currently active event loop.
  *
@@ -147,6 +150,16 @@ function getStdinBuffer(): string
     return getStdinLineReader()->getBuffer();
 }
 
+/**
+ * Clear data eventually buffered by the STDIN LineReader
+ *
+ * @return void
+ */
+function clearStdinBuffer()
+{
+    return getStdinLineReader()->clearBuffer();
+}
+
 /**
  * The STDOUT stream for the process associated with the currently active event loop.
  *
@@ -185,7 +198,6 @@ function getStderr(): ResourceOutputStream
     return $stream;
 }
 
-
 /**
  * Buffered async version of the readline() function.
  *
@@ -217,3 +229,30 @@ function bufferEcho($string): Promise
 {
     return getOutputBufferStream()->write($string);
 }
+function parseLineDelimitedJson(InputStream $stream, bool $assoc = false, int $depth = 512, int $options = 0): Iterator
+{
+    return new Producer(static function (callable $emit) use ($stream, $assoc, $depth, $options) {
+        $reader = new LineReader($stream);
+
+        while (null !== $line = yield $reader->readLine()) {
+            $line = \trim($line);
+
+            if ($line === '') {
+                continue;
+            }
+
+            /** @noinspection PhpComposerExtensionStubsInspection */
+            $data = \json_decode($line, $assoc, $depth, $options);
+            /** @noinspection PhpComposerExtensionStubsInspection */
+            $error = \json_last_error();
+
+            /** @noinspection PhpComposerExtensionStubsInspection */
+            if ($error !== \JSON_ERROR_NONE) {
+                /** @noinspection PhpComposerExtensionStubsInspection */
+                throw new StreamException('Failed to parse JSON: ' . \json_last_error_msg(), $error);
+            }
+
+            yield $emit($data);
+        }
+    });
+}

test/LineReaderTest.php

@@ -59,6 +59,22 @@ class LineReaderTest extends TestCase
         $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) {

test/ParseLineDelimitedJsonTest.php

@@ -0,0 +1,38 @@
+<?php
+
+/** @noinspection PhpComposerExtensionStubsInspection */
+
+/** @noinspection PhpUnhandledExceptionInspection */
+
+namespace Amp\ByteStream\Test;
+
+use Amp\ByteStream\InMemoryStream;
+use Amp\ByteStream\StreamException;
+use Amp\Iterator;
+use Amp\PHPUnit\TestCase;
+use function Amp\ByteStream\parseLineDelimitedJson;
+use function Amp\Promise\wait;
+
+class ParseLineDelimitedJsonTest extends TestCase
+{
+    public function test()
+    {
+        $result = wait(Iterator\toArray(parseLineDelimitedJson(new InMemoryStream(\implode("\n", [
+            \json_encode(['foo' => "\nbar\r\n"]),
+            \json_encode(['foo' => []]),
+        ])))));
+
+        self::assertEquals([
+            (object) ['foo' => "\nbar\r\n"],
+            (object) ['foo' => []],
+        ], $result);
+    }
+
+    public function testInvalidJson()
+    {
+        $this->expectException(StreamException::class);
+        $this->expectExceptionMessage('Failed to parse JSON');
+
+        wait(Iterator\toArray(parseLineDelimitedJson(new InMemoryStream('{'))));
+    }
+}