Flow Http Stream DX improvements (#1497)

* Flow Http Stream DX improvements

* Updated DSL definitions

* CS Fixes

Author: norberttech

composer.json

@@ -152,7 +152,8 @@
             "src/lib/filesystem/src/Flow/Filesystem/DSL/functions.php",
             "src/lib/parquet/src/Flow/Parquet/functions.php",
             "src/lib/parquet/src/stubs.php",
-            "src/lib/snappy/polyfill.php"
+            "src/lib/snappy/polyfill.php",
+            "src/bridge/symfony/http-foundation/src/Flow/Bridge/Symfony/HttpFoundation/functions.php"
         ]
     },
     "autoload-dev": {

documentation/components/bridges/symfony-http-foundation-bridge.md

@@ -16,7 +16,7 @@ Http Foundation Bridge provides seamless integration between Symfony Http Founda
 ## Installation
 
 ```
-composer require flow-php/symfony-http-foundation-bridge
+composer require flow-php/symfony-http-foundation-bridge:1.x-dev
 ```
 
 ## Usage
@@ -30,87 +30,67 @@ files that normally would not fit in memory.
 
 namespace Symfony\Application\Controller;
 
-use Flow\Bridge\Symfony\HttpFoundation\DataStream;
-use Flow\Bridge\Symfony\HttpFoundation\Output\CSVOutput;
+use Flow\Bridge\Symfony\HttpFoundation\Response\FlowBufferedResponse;
+use Flow\Bridge\Symfony\HttpFoundation\Response\FlowStreamedResponse;
+use Flow\ETL\Transformation\AddRowIndex;
+use Flow\ETL\Transformation\Limit;
+use Flow\ETL\Transformation\MaskColumns;
 use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
-use Symfony\Component\HttpFoundation\Response;
 use Symfony\Component\Routing\Attribute\Route;
-use Flow\Bridge\Symfony\HttpFoundation\Transformation\MaskColumns;
-use Flow\Bridge\Symfony\HttpFoundation\Transformation\AddRowIndex;
+use function Flow\Bridge\Symfony\HttpFoundation\http_csv_output;
+use function Flow\Bridge\Symfony\HttpFoundation\http_stream_open;
 use function Flow\ETL\Adapter\Parquet\from_parquet;
 
 final class ReportsController extends AbstractController
 {
-    #[Route('/stream/report', name: 'stream-report')]
-    public function streamReport() : Response
+    #[Route('/report/stream', name: 'report_stream')]
+    public function streamReport() : FlowStreamedResponse
     {
-        return DataStream
-            ::open(from_parquet(__DIR__ . '/reports/orders.parquet'))
-            ->underFilename('orders.csv')
+        return http_stream_open(from_parquet(__DIR__ . '/reports/orders.parquet'))
+            ->headers(['X-Custom-Header' => 'Custom Value'])
             ->transform(
                 new MaskColumns(['email', 'address']),
                 new AddRowIndex()
             )
-            ->to(new CSVOutput(withHeader: true));
+            ->as('orders.csv')
+            ->status(200)
+            ->streamedResponse(http_csv_output());
+    }
+
+    #[Route('/report', name: 'report')]
+    public function bufferReport() : FlowBufferedResponse
+    {
+        return http_stream_open(from_parquet(__DIR__ . '/reports/orders.parquet'))
+            ->transform(
+                new Limit(100),
+                new MaskColumns(['email', 'address']),
+                new AddRowIndex(),
+            )
+            ->as('orders.csv')
+            ->response(http_csv_output());
     }
 }
 ```
 
 ## Available Outputs
 
-- `Flow\Bridge\Symfony\HttpFoundation\Output\CSVOutput` - converts dataset to CSV format.
-- `Flow\Bridge\Symfony\HttpFoundation\Output\JSONOutput` - converts dataset to JSON format.
-- `Flow\Bridge\Symfony\HttpFoundation\Output\ParquetOutput` - converts dataset to Parquet format.
-- `Flow\Bridge\Symfony\HttpFoundation\Output\XMLOutput` - converts dataset to XML format.
+- `Flow\Bridge\Symfony\HttpFoundation\Output\CSVOutput` - `http_csv_output()` - converts dataset to CSV format.
+- `Flow\Bridge\Symfony\HttpFoundation\Output\JSONOutput` - `http_json_output()` -converts dataset to JSON format.
+- `Flow\Bridge\Symfony\HttpFoundation\Output\ParquetOutput` - `http_parquet_output()` -converts dataset to Parquet format.
+- `Flow\Bridge\Symfony\HttpFoundation\Output\XMLOutput` - `http_xml_output()` -converts dataset to XML format.
 
 ## Modify output on the fly
 
 Sometimes we need to modify the output on the fly. 
-To do that, FlowStreamedResponse allows to pass a Transformation that will be applied on the dataset.
+To do that, FlowStreamedResponse allows passing a Transformation that will be applied on the dataset.
 
 ```php
-return new FlowStreamedResponse(
-    new ParquetEtractor(__DIR__ . '/reports/orders.parquet'),
-    new CSVOutput(withHeader: true),
-    new class implements Transformation {
-        public function transform(DataFrame $dataFrame): DataFrame
-        {
-            return $dataFrame->withColumn('time', \time());
-        }
+new class implements Transformation {
+    public function transform(DataFrame $dataFrame): DataFrame
+    {
+        return $dataFrame->withColumn('time', \time());
     }
-);
+}
 ```
 
 Above example will add a new column `time` to the dataset with the current timestamp.
-
-Predefined Transformations: 
-
-- `Flow\Bridge\Symfony\HttpFoundation\Transformation\MaskColumns` - mask columns with `*****` value.
-
-```php
-<?php
-
-declare(strict_types=1);
-
-namespace Symfony\Application\Controller;
-
-use Flow\Bridge\Symfony\HttpFoundation\FlowStreamedResponse;
-use Flow\Bridge\Symfony\HttpFoundation\Output\CSVOutput;
-use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
-use Symfony\Component\HttpFoundation\Response;
-use Symfony\Component\Routing\Attribute\Route;
-use function Flow\ETL\Adapter\Parquet\ParquetEtractor;
-
-final class ReportsController extends AbstractController
-{
-    #[Route('/stream/report', name: 'stream-report')]
-    public function streamReport() : Response
-    {
-        return new FlowStreamedResponse(
-            new ParquetEtractor(__DIR__ . '/reports/orders.parquet'),
-            new CSVOutput(withHeader: true),
-            new MaskColumns(['email', 'address'])
-        );
-    }
-}
-```

src/adapter/etl-adapter-parquet/tests/Flow/ETL/Adapter/Parquet/Tests/Integration/ParquetTest.php

@@ -6,50 +6,38 @@
 
 use function Flow\ETL\Adapter\Parquet\{from_parquet, to_parquet};
 use function Flow\ETL\DSL\data_frame;
-use function Flow\ETL\DSL\{df, from_array, json_schema, schema, str_schema};
+use function Flow\ETL\DSL\{config, from_array, json_schema, schema, str_schema};
+use function Flow\Filesystem\DSL\{path};
 use Flow\ETL\Tests\Double\FakeExtractor;
 use Flow\ETL\{Tests\FlowTestCase};
-use Flow\Parquet\ParquetFile\Compressions;
-use Flow\Parquet\Reader;
 use Ramsey\Uuid\Uuid;
 
 final class ParquetTest extends FlowTestCase
 {
-    public function test_writing_to_file() : void
+    public function test_writing_and_reading_into_parquet() : void
     {
-        $path = __DIR__ . '/var/file.snappy.parquet';
-        $this->removeFile($path);
+        $path = path('memory://var/file.snappy.parquet');
 
-        df()
+        $config = config();
+        data_frame($config)
             ->read(new FakeExtractor(10))
             ->drop('null', 'array', 'object', 'enum')
             ->write(to_parquet($path))
             ->run();
 
         self::assertEquals(
             10,
-            (data_frame())
+            (data_frame($config))
                 ->read(from_parquet($path))
                 ->count()
         );
-
-        $parquetFile = (new Reader())->read($path);
-        self::assertNotEmpty($parquetFile->metadata()->columnChunks());
-
-        foreach ($parquetFile->metadata()->columnChunks() as $columnChunk) {
-            self::assertSame(Compressions::SNAPPY, $columnChunk->codec());
-        }
-
-        self::assertFileExists($path);
-        $this->removeFile($path);
     }
 
     public function test_writing_with_provided_schema() : void
     {
-        $path = __DIR__ . '/var/file_schema.snappy.parquet';
-        $this->removeFile($path);
-
-        df()
+        $path = path('memory://var/file_schema.snappy.parquet');
+        $config = config();
+        data_frame($config)
             ->read(from_array([
                 ['id' => 1, 'name' => 'test', 'uuid' => Uuid::fromString('26fd21b0-6080-4d6c-bdb4-1214f1feffef'), 'json' => '[{"id":1,"name":"test"},{"id":2,"name":"test"}]'],
                 ['id' => 2, 'name' => 'test', 'uuid' => Uuid::fromString('26fd21b0-6080-4d6c-bdb4-1214f1feffef'), 'json' => '[{"id":1,"name":"test"},{"id":2,"name":"test"}]'],
@@ -69,14 +57,13 @@ public function test_writing_with_provided_schema() : void
                 ['id' => '1', 'name' => 'test', 'uuid' => new \Flow\ETL\PHP\Value\Uuid('26fd21b0-6080-4d6c-bdb4-1214f1feffef'), 'json' => [['id' => 1, 'name' => 'test'], ['id' => 2, 'name' => 'test']]],
                 ['id' => '2', 'name' => 'test', 'uuid' => new \Flow\ETL\PHP\Value\Uuid('26fd21b0-6080-4d6c-bdb4-1214f1feffef'), 'json' => [['id' => 1, 'name' => 'test'], ['id' => 2, 'name' => 'test']]],
             ],
-            df()
+            data_frame($config)
                 ->read(from_parquet($path))
                 ->fetch()
                 ->toArray()
         );
 
-        self::assertFileExists($path);
-        $this->removeFile($path);
+        self::assertTrue($config->fstab()->for($path)->status($path)?->isFile());
     }
 
     /**

src/bridge/symfony/http-foundation/composer.json

@@ -24,6 +24,9 @@
     },
     "license": "MIT",
     "autoload": {
+        "files": [
+            "src/Flow/Bridge/Symfony/HttpFoundation/functions.php"
+        ],
         "psr-4": {
             "Flow\\": [
                 "src/Flow"

src/bridge/symfony/http-foundation/src/Flow/Bridge/Symfony/HttpFoundation/DataStream.php

@@ -4,6 +4,7 @@
 
 namespace Flow\Bridge\Symfony\HttpFoundation;
 
+use Flow\Bridge\Symfony\HttpFoundation\Response\{FlowBufferedResponse, FlowStreamedResponse};
 use Flow\ETL\{Extractor, Transformation, Transformations};
 use Symfony\Component\HttpFoundation\{HeaderUtils, Response};
 
@@ -21,8 +22,6 @@ final class DataStream
         'Pragma' => 'no-cache', // Backward compatibility for HTTP/1.0
     ];
 
-    private ?Output $output = null;
-
     private int $status = Response::HTTP_OK;
 
     /**
@@ -40,59 +39,86 @@ public static function open(Extractor $extractor) : self
     }
 
     /**
-     * Send the data stream to the output.
+     * Set the filename for the response.
+     * If the attachment flag is set to true, the response will be treated as an attachment meaning that
+     * the browser will prompt the user to download the file.
      */
-    public function sendTo(Output $output) : FlowStreamedResponse
+    public function as(string $name, bool $attachment = true) : self
     {
-        $this->output = $output;
+        $this->headers['Content-Disposition'] = HeaderUtils::makeDisposition(
+            $attachment ? HeaderUtils::DISPOSITION_ATTACHMENT : HeaderUtils::DISPOSITION_INLINE,
+            $name
+        );
 
-        $this->headers['Content-Type'] = $this->output->type()->toContentTypeHeader();
+        return $this;
+    }
 
-        return new FlowStreamedResponse(
+    /**
+     * Set additional headers.
+     * Headers are merged with the default headers.
+     */
+    public function headers(array $headers) : self
+    {
+        $this->headers = array_merge($this->headers, $headers);
+
+        return $this;
+    }
+
+    /**
+     * Create regular response where whole dataset is loaded into the memory.
+     * It's highly recommended to use limit transformation to avoid loading entire dataset into the memory.
+     * Some extractors like Parquet/Elasticsearch/Doctrine allows also for setting offset directly on the extractor.
+     */
+    public function response(Output $output) : FlowBufferedResponse
+    {
+        $this->headers['Content-Type'] = $output->type()->toContentTypeHeader();
+
+        return new FlowBufferedResponse(
             $this->extractor,
-            $this->output,
+            $output,
             \count($this->transformations) ? new Transformations(...$this->transformations) : new Transformations(),
             $this->status,
             $this->headers
         );
     }
 
     /**
-     * Apply transformations to the data stream.
-     * Transformations are applied in the order they are passed.
-     * Transformations are applied on the fly, while streaming the data, this means
-     * that any resource expensive transformations like for example aggregations or sorting
-     * might significantly slow down the streaming process or even cause out of memory errors.
+     * Set the HTTP status code. Default is 200.
      */
-    public function transform(Transformation ...$transformations) : self
+    public function status(int $status) : self
     {
-        $this->transformations = $transformations;
+        $this->status = $status;
 
         return $this;
     }
 
     /**
-     * Set the filename for the response.
-     * If the attachment flag is set to true, the response will be treated as an attachment meaning that
-     * the browser will prompt the user to download the file.
+     * Send the data stream to the output.
      */
-    public function underFilename(string $name, bool $attachment = true) : self
+    public function streamedResponse(Output $output) : FlowStreamedResponse
     {
-        $this->headers['Content-Disposition'] = HeaderUtils::makeDisposition(
-            $attachment ? HeaderUtils::DISPOSITION_ATTACHMENT : HeaderUtils::DISPOSITION_INLINE,
-            $name
-        );
 
-        return $this;
+        $this->headers['Content-Type'] = $output->type()->toContentTypeHeader();
+
+        return new FlowStreamedResponse(
+            $this->extractor,
+            $output,
+            \count($this->transformations) ? new Transformations(...$this->transformations) : new Transformations(),
+            $this->status,
+            $this->headers
+        );
     }
 
     /**
-     * Set additional headers.
-     * Headers are merged with the default headers.
+     * Apply transformations to the data stream.
+     * Transformations are applied in the order they are passed.
+     * Transformations are applied on the fly, while streaming the data, this means
+     * that any resource expensive transformations like for example aggregations or sorting
+     * might significantly slow down the streaming process or even cause out of memory errors.
      */
-    public function withHeaders(array $headers) : self
+    public function transform(Transformation ...$transformations) : self
     {
-        $this->headers = array_merge($this->headers, $headers);
+        $this->transformations = $transformations;
 
         return $this;
     }
@@ -109,14 +135,4 @@ public function withoutHeader(string $name) : self
 
         return $this;
     }
-
-    /**
-     * Set the HTTP status code. Default is 200.
-     */
-    public function withStatus(int $status) : self
-    {
-        $this->status = $status;
-
-        return $this;
-    }
 }

src/bridge/symfony/http-foundation/src/Flow/Bridge/Symfony/HttpFoundation/Output.php

@@ -9,7 +9,9 @@
 
 interface Output
 {
-    public function loader() : Loader;
+    public function memoryLoader(string $id) : Loader;
+
+    public function stdoutLoader() : Loader;
 
     public function type() : Type;
 }