Feat: STDIN Extractor / STDOUT Loader (#35)

README.md

@@ -26,6 +26,7 @@ Table of Contents
     - [Difference between yield and return in transformers](doc/advanced_usage.md#difference-between-yield-and-return-in-transformers)
     - [Next tick](doc/advanced_usage.md#next-tick)
     - [Chaining extractors / transformers / loaders](doc/advanced_usage.md#chaining-extractors--transformers--loaders)
+    - [Reading from STDIN / Writing to STDOUT](doc/advanced_usage.md#reading-from-stdin--writing-to-stdout)
     - [Instantiators](doc/advanced_usage.md#instantiators)
 - [Recipes](doc/recipes.md)
 - [Contributing](#contribute)

doc/advanced_usage.md

@@ -116,6 +116,22 @@ var_dump([...$a]); // ['F-O-O', 'B-A-R']
 var_dump([...$b]); // ['F-O-O', 'B-A-R']
 ```
 
+Reading from STDIN / Writing to STDOUT
+--------------------------------------
+
+Easy as hell.
+
+```php
+use function BenTools\ETL\stdIn;
+use function BenTools\ETL\stdOut;
+use function BenTools\ETL\transformWith;
+
+transformWith(fn (string $line) => strtoupper($line))
+    ->extractFrom(stdIn())
+    ->loadInto(stdOut())
+    ->process();
+```
+
 Recipes
 -------
 

src/Extractor/STDINExtractor.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BenTools\ETL\Extractor;
+
+use BenTools\ETL\EtlState;
+use Iterator;
+use SplFileObject;
+
+/**
+ * @implements Iterator<int, string>
+ */
+final class STDINExtractor implements Iterator, ExtractorInterface
+{
+    private SplFileObject $stdIn;
+
+    public function current(): string|false
+    {
+        return $this->stdIn->current();
+    }
+
+    public function next(): void
+    {
+        $this->stdIn->next();
+    }
+
+    public function key(): int
+    {
+        return $this->stdIn->key();
+    }
+
+    public function valid(): bool
+    {
+        return $this->stdIn->valid();
+    }
+
+    public function rewind(): void
+    {
+        $this->stdIn = new SplFileObject('php://stdin');
+        $this->stdIn->setFlags(SplFileObject::DROP_NEW_LINE);
+    }
+
+    public function extract(EtlState $state): iterable
+    {
+        yield from $this;
+    }
+}

src/Loader/STDOUTLoader.php

@@ -0,0 +1,55 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BenTools\ETL\Loader;
+
+use BenTools\ETL\EtlState;
+use BenTools\ETL\Exception\LoadException;
+
+use function fclose;
+use function fopen;
+use function fwrite;
+use function get_debug_type;
+use function is_string;
+use function sprintf;
+
+use const PHP_EOL;
+
+final readonly class STDOUTLoader implements LoaderInterface
+{
+    public function __construct(
+        private string $eol = PHP_EOL,
+    ) {
+    }
+
+    public function load(mixed $item, EtlState $state): void
+    {
+        if (!is_string($item)) {
+            throw new LoadException(sprintf('Expected string, got %s.', get_debug_type($item)));
+        }
+
+        $state->context[__CLASS__]['pending'][] = $item;
+    }
+
+    public function flush(bool $isPartial, EtlState $state): int
+    {
+        $pendingItems = $state->context[__CLASS__]['pending'] ?? [];
+        $state->context[__CLASS__]['resource'] ??= fopen('php://stdout', 'wb+');
+        $state->context[__CLASS__]['nbWrittenBytes'] ??= 0;
+        foreach ($pendingItems as $item) {
+            $state->context[__CLASS__]['nbWrittenBytes'] += fwrite(
+                $state->context[__CLASS__]['resource'],
+                $item.$this->eol,
+            );
+        }
+
+        $nbWrittenBytes = $state->context[__CLASS__]['nbWrittenBytes'];
+        if (!$isPartial) {
+            // fclose($state->context[__CLASS__]['resource']);
+            unset($state->context[__CLASS__]);
+        }
+
+        return $nbWrittenBytes;
+    }
+}

src/functions.php

@@ -6,9 +6,11 @@
 
 use BenTools\ETL\Extractor\ChainExtractor;
 use BenTools\ETL\Extractor\ExtractorInterface;
+use BenTools\ETL\Extractor\STDINExtractor;
 use BenTools\ETL\Internal\Ref;
 use BenTools\ETL\Loader\ChainLoader;
 use BenTools\ETL\Loader\LoaderInterface;
+use BenTools\ETL\Loader\STDOUTLoader;
 use BenTools\ETL\Recipe\Recipe;
 use BenTools\ETL\Transformer\ChainTransformer;
 use BenTools\ETL\Transformer\TransformerInterface;
@@ -93,3 +95,13 @@ function chain(ExtractorInterface|TransformerInterface|LoaderInterface $service,
         $service instanceof LoaderInterface => ChainLoader::from($service),
     };
 }
+
+function stdIn(): STDINExtractor
+{
+    return new STDINExtractor();
+}
+
+function stdOut(): STDOUTLoader
+{
+    return new STDOUTLoader();
+}

tests/Behavior/FlushTest.php

@@ -7,7 +7,7 @@
 use BenTools\ETL\EtlConfiguration;
 use BenTools\ETL\EtlExecutor;
 use BenTools\ETL\EventDispatcher\Event\ExtractEvent;
-use BenTools\ETL\Tests\InMemoryLoader;
+use BenTools\ETL\Tests\Stubs\InMemoryLoader;
 
 use function expect;
 

tests/Behavior/NextTickTest.php

@@ -10,7 +10,7 @@
 use BenTools\ETL\EtlState;
 use BenTools\ETL\EventDispatcher\Event\ExtractEvent;
 use BenTools\ETL\EventDispatcher\Event\LoadEvent;
-use BenTools\ETL\Tests\InMemoryLoader;
+use BenTools\ETL\Tests\Stubs\InMemoryLoader;
 
 use function expect;
 

tests/Behavior/SkipTest.php

@@ -13,7 +13,7 @@
 use function expect;
 
 it('skips items during extraction', function () {
-    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/data/10-biggest-cities.csv', [
+    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/Data/10-biggest-cities.csv', [
         'columns' => 'auto',
     ]);
     $cities = [];
@@ -49,7 +49,7 @@
 });
 
 it('skips items during transformation', function () {
-    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/data/10-biggest-cities.csv', [
+    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/Data/10-biggest-cities.csv', [
         'columns' => 'auto',
     ]);
     $cities = [];

tests/Behavior/StopTest.php

@@ -14,7 +14,7 @@
 use function expect;
 
 it('stops the process during extraction', function () {
-    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/data/10-biggest-cities.csv', [
+    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/Data/10-biggest-cities.csv', [
         'columns' => 'auto',
     ]);
     $cities = [];
@@ -44,7 +44,7 @@
 });
 
 it('stops the process during transformation', function () {
-    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/data/10-biggest-cities.csv', [
+    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/Data/10-biggest-cities.csv', [
         'columns' => 'auto',
     ]);
     $cities = [];
@@ -75,7 +75,7 @@
 });
 
 it('stops the process during loading', function () {
-    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/data/10-biggest-cities.csv', [
+    $extractor = new CSVExtractor('file://'.dirname(__DIR__).'/Data/10-biggest-cities.csv', [
         'columns' => 'auto',
     ]);
     $cities = [];

tests/InMemoryLoader.php → tests/Stubs/InMemoryLoader.php

@@ -2,7 +2,7 @@
 
 declare(strict_types=1);
 
-namespace BenTools\ETL\Tests;
+namespace BenTools\ETL\Tests\Stubs;
 
 use BenTools\ETL\EtlState;
 use BenTools\ETL\Loader\LoaderInterface;

tests/Stubs/STDINStub.php

@@ -0,0 +1,92 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BenTools\ETL\Tests\Stubs;
+
+use function file_exists;
+use function file_put_contents;
+use function min;
+use function stream_wrapper_register;
+use function stream_wrapper_restore;
+use function stream_wrapper_unregister;
+use function strlen;
+use function substr;
+
+/**
+ * Inspired by @KEINOS.
+ *
+ * @see https://github.com/KEINOS/Practice_PHPUnit-test-of-STDIN
+ */
+final class STDINStub
+{
+    private string $bufferFilename;
+    private int $index;
+    private int $length;
+    private string $data = '';
+    public mixed $context;
+
+    public function __construct()
+    {
+        $this->bufferFilename = sys_get_temp_dir().DIRECTORY_SEPARATOR.'php_input.txt';
+        $this->index = 0;
+        if (file_exists($this->bufferFilename)) {
+            $this->data = file_get_contents($this->bufferFilename);
+        }
+        $this->length = strlen($this->data);
+    }
+
+    public function stream_open(): true
+    {
+        return true;
+    }
+
+    public function url_stat(): false
+    {
+        return false;
+    }
+
+    public function stream_close(): void
+    {
+    }
+
+    public function stream_stat(): false
+    {
+        return false;
+    }
+
+    public function stream_flush(): true
+    {
+        return true;
+    }
+
+    public function stream_read(int $count): string
+    {
+        $length = min($count, $this->length - $this->index);
+        $data = substr($this->data, $this->index);
+        $this->index += $length;
+
+        return $data;
+    }
+
+    public function stream_eof(): bool
+    {
+        return $this->index >= $this->length;
+    }
+
+    public function stream_write(string $data): false|int
+    {
+        return file_put_contents($this->bufferFilename, $data);
+    }
+
+    public static function emulate(string $stdInContent, callable $beforeRestore): mixed
+    {
+        stream_wrapper_unregister('php');
+        stream_wrapper_register('php', __CLASS__);
+        file_put_contents('php://stdin', $stdInContent);
+        $result = $beforeRestore();
+        stream_wrapper_restore('php');
+
+        return $result;
+    }
+}

tests/Stubs/STDOUTStub.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BenTools\ETL\Tests\Stubs;
+
+use function fopen;
+use function stream_bucket_append;
+use function stream_bucket_make_writeable;
+use function stream_filter_append;
+use function stream_filter_register;
+
+final class STDOUTStub
+{
+    public string $filtername = 'intercept';
+    public ?array $params = null; // @phpstan-ignore-line
+    private static string $storage = '';
+
+    // @phpstan-ignore-next-line
+    public function filter($in, $out, &$consumed, bool $closing): int
+    {
+        while ($bucket = stream_bucket_make_writeable($in)) {
+            self::$storage .= $bucket->data;
+            $consumed += $bucket->datalen;
+            stream_bucket_append($out, $bucket);
+        }
+
+        return PSFS_PASS_ON;
+    }
+
+    public static function read(): string
+    {
+        return self::$storage;
+    }
+
+    public static function emulate(callable $beforeRestore, string $filename = 'php://stdout'): string
+    {
+        stream_filter_register('intercept', __CLASS__);
+        $stdout = fopen($filename, 'wb+');
+        $filter = stream_filter_append($stdout, 'intercept');
+        $beforeRestore($stdout);
+        $result = self::$storage;
+
+        self::$storage = '';
+
+        return $result;
+    }
+}

tests/Unit/Extractor/CSVExtractorTest.php

@@ -21,8 +21,8 @@
 
 it('iterates over a string containing CSV data', function () {
     $state = new EtlState();
-    $content = file_get_contents(dirname(__DIR__, 2).'/data/10-biggest-cities.csv');
-    $expected = require dirname(__DIR__, 2).'/data/10-biggest-cities.php';
+    $content = file_get_contents(dirname(__DIR__, 2).'/Data/10-biggest-cities.csv');
+    $expected = require dirname(__DIR__, 2).'/Data/10-biggest-cities.php';
     $extractor = new CSVExtractor($content, ['columns' => 'auto']);
 
     // When
@@ -36,7 +36,7 @@
     $extractor = new CSVExtractor(options: ['columns' => 'auto']);
 
     // When
-    $state = new EtlState(source: 'file://'.dirname(__DIR__, 2).'/data/10-biggest-cities.csv');
+    $state = new EtlState(source: 'file://'.dirname(__DIR__, 2).'/Data/10-biggest-cities.csv');
     $extractedItems = [...$extractor->extract($state)];
 
     // Then