Json Loader - allow to write each node in a new line (#1183)

src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/JsonLoader.php

@@ -18,7 +18,9 @@ final class JsonLoader implements Closure, Loader, Loader\FileLoader
 
     public function __construct(
         private readonly Path $path,
-        private readonly string $dateTimeFormat = \DateTimeInterface::ATOM
+        private readonly int $flats = JSON_THROW_ON_ERROR,
+        private readonly string $dateTimeFormat = \DateTimeInterface::ATOM,
+        private readonly bool $putRowsInNewLines = false
     ) {
         if ($this->path->isPattern()) {
             throw new \InvalidArgumentException("JsonLoader path can't be pattern, given: " . $this->path->path());
@@ -29,7 +31,7 @@ public function closure(FlowContext $context) : void
     {
         foreach ($context->streams() as $stream) {
             if ($stream->path()->extension() === 'json') {
-                $stream->append(']');
+                $stream->append($this->putRowsInNewLines ? "\n]" : ']');
             }
         }
 
@@ -65,7 +67,7 @@ public function write(Rows $nextRows, array $partitions, FlowContext $context) :
                 $this->writes[$stream->path()->path()] = 0;
             }
 
-            $stream->append('[');
+            $stream->append($this->putRowsInNewLines ? "[\n" : '[');
         } else {
             $stream = $streams->writeTo($this->path, $partitions);
         }
@@ -86,9 +88,16 @@ public function writeJSON(Rows $rows, DestinationStream $stream, RowsNormalizer
             return;
         }
 
+        $separator = $this->putRowsInNewLines ? ",\n" : ',';
+
         foreach ($normalizer->normalize($rows) as $normalizedRow) {
-            $json = json_encode($normalizedRow, JSON_THROW_ON_ERROR);
-            $json = ($this->writes[$stream->path()->path()] > 0) ? ',' . $json : $json;
+            $json = json_encode($normalizedRow, $this->flats);
+
+            if ($json === false) {
+                throw new RuntimeException('Failed to encode JSON: ' . json_last_error_msg());
+            }
+
+            $json = ($this->writes[$stream->path()->path()] > 0) ? ($separator . $json) : $json;
 
             $stream->append($json);
         }

src/adapter/etl-adapter-json/src/Flow/ETL/Adapter/JSON/functions.php

@@ -47,7 +47,11 @@ function from_json(
  *
  * @return Loader
  */
-function to_json(string|Path $path, string $date_time_format = \DateTimeInterface::ATOM) : Loader
-{
-    return new JsonLoader(\is_string($path) ? Path::realpath($path) : $path, $date_time_format);
+function to_json(
+    string|Path $path,
+    int $flags = JSON_THROW_ON_ERROR,
+    string $date_time_format = \DateTimeInterface::ATOM,
+    bool $put_rows_in_new_lines = false
+) : Loader {
+    return new JsonLoader(\is_string($path) ? Path::realpath($path) : $path, $flags, $date_time_format, $put_rows_in_new_lines);
 }

src/adapter/etl-adapter-json/tests/Flow/ETL/Adapter/JSON/Tests/Integration/JsonTest.php

@@ -6,7 +6,7 @@
 
 use function Flow\ETL\Adapter\JSON\from_json;
 use function Flow\ETL\Adapter\Json\to_json;
-use function Flow\ETL\DSL\{df, overwrite};
+use function Flow\ETL\DSL\{df, from_array, overwrite};
 use function Flow\Filesystem\DSL\path;
 use Flow\ETL\Adapter\JSON\JsonLoader;
 use Flow\ETL\Tests\Double\FakeExtractor;
@@ -80,4 +80,54 @@ public function test_json_loader_overwrite_mode() : void
             \unlink($path);
         }
     }
+
+    public function test_putting_each_row_in_a_new_line() : void
+    {
+        df()
+            ->read(from_array([
+                ['name' => 'John', 'age' => 30],
+                ['name' => 'Jane', 'age' => 25],
+            ]))
+            ->saveMode(overwrite())
+            ->write(to_json($path = __DIR__ . '/var/test_putting_each_row_in_a_new_line.json', put_rows_in_new_lines: true))
+            ->run();
+
+        self::assertStringContainsString(
+            <<<'JSON'
+[
+{"name":"John","age":30},
+{"name":"Jane","age":25}
+]
+JSON,
+            \file_get_contents($path)
+        );
+    }
+
+    public function test_putting_each_row_in_a_new_line_with_json_pretty_print_flag() : void
+    {
+        df()
+            ->read(from_array([
+                ['name' => 'John', 'age' => 30],
+                ['name' => 'Jane', 'age' => 25],
+            ]))
+            ->saveMode(overwrite())
+            ->write(to_json($path = __DIR__ . '/var/test_putting_each_row_in_a_new_line.json', flags: JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT, put_rows_in_new_lines: true))
+            ->run();
+
+        self::assertStringContainsString(
+            <<<'JSON'
+[
+{
+    "name": "John",
+    "age": 30
+},
+{
+    "name": "Jane",
+    "age": 25
+}
+]
+JSON,
+            \file_get_contents($path)
+        );
+    }
 }