Add line delimited JSON parser (#65)

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.
  *
@@ -155,3 +158,31 @@ function getStderr(): ResourceOutputStream
 
     return $stream;
 }
+
+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/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('{'))));
+    }
+}