Feat: React streams support (#37)

README.md

@@ -28,6 +28,7 @@ Table of Contents
     - [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)
+    - [Using React Streams (ReactPHP support)](doc/advanced_usage.md#using-react-streams-experimental)
 - [Recipes](doc/recipes.md)
 - [Contributing](#contribute)
 - [License](#license)

composer.json

@@ -18,6 +18,7 @@
     "pestphp/pest": "^2.24",
     "phpstan/phpstan": "^1.10",
     "phpstan/phpstan-mockery": "^1.1",
+    "react/stream": "^1.3",
     "symfony/var-dumper": "*"
   },
   "license": "MIT",

doc/advanced_usage.md

@@ -158,3 +158,41 @@ $report = withRecipe(new LoggerRecipe($logger))
     ->transformWith(fn ($value) => strtoupper($value))
     ->process(['foo', 'bar']);
 ```
+
+Using React streams (experimental)
+----------------------------------
+
+You can plug your ETL dataflows to any [React Stream](https://github.com/reactphp/stream).
+
+Example with a TCP server:
+
+```php
+use BenTools\ETL\EtlConfiguration;
+use BenTools\ETL\EventDispatcher\Event\ExtractEvent;
+use BenTools\ETL\EventDispatcher\Event\InitEvent;
+use React\Socket\ConnectionInterface;
+use React\Socket\SocketServer;
+
+use function BenTools\ETL\stdOut;
+use function BenTools\ETL\useReact;
+
+$socket = new SocketServer('127.0.0.1:7000');
+
+$etl = useReact() // or (new EtlExecutor())->withRecipe(new ReactStreamProcessor());
+    ->loadInto(stdOut())
+    ->onInit(function (InitEvent $event) {
+        /** @var ConnectionInterface $stream */
+        $stream = $event->state->source;
+        $stream->on('close', function () use ($event) {
+            $event->state->stop(); // Will flush all pending items and gracefully stop the ETL for that connection
+        });
+    })
+    ->withOptions(new EtlConfiguration(flushEvery: 1)) // Optionally, flush on each data event
+    ->onExtract(function (ExtractEvent $event) {
+        if (!preg_match('//u', $event->item)) {
+            $event->state->skip(); // Ignore binary data
+        }
+    });
+
+$socket->on('connection', $etl->process(...));
+```

src/EtlExecutor.php

@@ -20,15 +20,15 @@
 use BenTools\ETL\Exception\StopRequest;
 use BenTools\ETL\Exception\TransformException;
 use BenTools\ETL\Extractor\ExtractorInterface;
-use BenTools\ETL\Extractor\ExtractorProcessorInterface;
 use BenTools\ETL\Extractor\IterableExtractor;
-use BenTools\ETL\Extractor\IterableExtractorProcessor;
 use BenTools\ETL\Internal\ClonableTrait;
 use BenTools\ETL\Internal\ConditionalLoaderTrait;
 use BenTools\ETL\Internal\EtlBuilderTrait;
 use BenTools\ETL\Internal\TransformResult;
 use BenTools\ETL\Loader\InMemoryLoader;
 use BenTools\ETL\Loader\LoaderInterface;
+use BenTools\ETL\Processor\IterableProcessor;
+use BenTools\ETL\Processor\ProcessorInterface;
 use BenTools\ETL\Transformer\NullTransformer;
 use BenTools\ETL\Transformer\TransformerInterface;
 use Psr\EventDispatcher\EventDispatcherInterface;
@@ -57,7 +57,7 @@ public function __construct(
         public readonly TransformerInterface $transformer = new NullTransformer(),
         public readonly LoaderInterface $loader = new InMemoryLoader(),
         public readonly EtlConfiguration $options = new EtlConfiguration(),
-        public readonly ExtractorProcessorInterface $processor = new IterableExtractorProcessor(),
+        public readonly ProcessorInterface $processor = new IterableProcessor(),
     ) {
         $this->listenerProvider = new PrioritizedListenerProvider();
         $this->eventDispatcher = new EventDispatcher($this->listenerProvider);

src/Extractor/ReactStreamExtractor.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BenTools\ETL\Extractor;
+
+use BenTools\ETL\EtlState;
+use React\Stream\ReadableStreamInterface;
+
+final readonly class ReactStreamExtractor implements ExtractorInterface
+{
+    public function __construct(
+        public ?ReadableStreamInterface $stream = null,
+    ) {
+    }
+
+    public function extract(EtlState $state): ReadableStreamInterface
+    {
+        return $state->source ?? $this->stream;
+    }
+}

src/Internal/EtlBuilderTrait.php

@@ -8,10 +8,10 @@
 use BenTools\ETL\Extractor\CallableExtractor;
 use BenTools\ETL\Extractor\ChainExtractor;
 use BenTools\ETL\Extractor\ExtractorInterface;
-use BenTools\ETL\Extractor\ExtractorProcessorInterface;
 use BenTools\ETL\Loader\CallableLoader;
 use BenTools\ETL\Loader\ChainLoader;
 use BenTools\ETL\Loader\LoaderInterface;
+use BenTools\ETL\Processor\ProcessorInterface;
 use BenTools\ETL\Recipe\Recipe;
 use BenTools\ETL\Transformer\CallableTransformer;
 use BenTools\ETL\Transformer\ChainTransformer;
@@ -103,7 +103,7 @@ public function withRecipe(Recipe|callable $recipe, Recipe|callable ...$recipes)
         return $executor;
     }
 
-    public function withProcessor(ExtractorProcessorInterface $processor): self
+    public function withProcessor(ProcessorInterface $processor): self
     {
         return $this->cloneWith(['processor' => $processor]);
     }

src/Iterator/CSVIterator.php

@@ -32,10 +32,11 @@
     private array $options;
 
     /**
+     * @param Traversable<string>                                                                                                                            $text
      * @param array{delimiter?: string, enclosure?: string, escapeString?: string, columns?: 'auto'|string[]|null, normalizers?: ValueNormalizerInterface[]} $options
      */
     public function __construct(
-        private PregSplitIterator|StrTokIterator|FileIterator $text,
+        private Traversable $text,
         array $options = [],
     ) {
         $resolver = (new OptionsResolver())->setIgnoreUndefined();

src/Extractor/IterableExtractorProcessor.php → src/Processor/IterableProcessor.php

@@ -2,7 +2,7 @@
 
 declare(strict_types=1);
 
-namespace BenTools\ETL\Extractor;
+namespace BenTools\ETL\Processor;
 
 use BenTools\ETL\EtlExecutor;
 use BenTools\ETL\EtlState;
@@ -16,7 +16,7 @@
 /**
  * @internal
  */
-final readonly class IterableExtractorProcessor implements ExtractorProcessorInterface
+final readonly class IterableProcessor implements ProcessorInterface
 {
     public function supports(mixed $extracted): bool
     {

src/Extractor/ExtractorProcessorInterface.php → src/Processor/ProcessorInterface.php

@@ -2,12 +2,12 @@
 
 declare(strict_types=1);
 
-namespace BenTools\ETL\Extractor;
+namespace BenTools\ETL\Processor;
 
 use BenTools\ETL\EtlExecutor;
 use BenTools\ETL\EtlState;
 
-interface ExtractorProcessorInterface
+interface ProcessorInterface
 {
     public function supports(mixed $extracted): bool;
 

src/Processor/ReactStreamProcessor.php

@@ -0,0 +1,61 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BenTools\ETL\Processor;
+
+use BenTools\ETL\EtlExecutor;
+use BenTools\ETL\EtlState;
+use BenTools\ETL\Exception\ExtractException;
+use BenTools\ETL\Exception\SkipRequest;
+use BenTools\ETL\Exception\StopRequest;
+use BenTools\ETL\Extractor\ReactStreamExtractor;
+use BenTools\ETL\Recipe\Recipe;
+use React\EventLoop\Loop;
+use React\Stream\ReadableStreamInterface;
+use Throwable;
+
+use function is_string;
+use function trim;
+
+/**
+ * @experimental
+ */
+final class ReactStreamProcessor extends Recipe implements ProcessorInterface
+{
+    public function supports(mixed $extracted): bool
+    {
+        return $extracted instanceof ReadableStreamInterface;
+    }
+
+    /**
+     * @param ReadableStreamInterface $stream
+     */
+    public function process(EtlExecutor $executor, EtlState $state, mixed $stream): EtlState
+    {
+        $key = -1;
+        $stream->on('data', function (mixed $item) use ($executor, &$key, $state, $stream) {
+            if (is_string($item)) {
+                $item = trim($item);
+            }
+            try {
+                $executor->processItem($item, ++$key, $state);
+            } catch (SkipRequest) {
+            } catch (StopRequest) {
+                $stream->close();
+            } catch (Throwable $e) {
+                $stream->close();
+                ExtractException::emit($executor, $e, $state);
+            }
+        });
+
+        Loop::run();
+
+        return $state->getLastVersion();
+    }
+
+    public function decorate(EtlExecutor $executor): EtlExecutor
+    {
+        return $executor->extractFrom(new ReactStreamExtractor())->withProcessor($this);
+    }
+}

src/functions.php

@@ -10,6 +10,7 @@
 use BenTools\ETL\Loader\ChainLoader;
 use BenTools\ETL\Loader\LoaderInterface;
 use BenTools\ETL\Loader\STDOUTLoader;
+use BenTools\ETL\Processor\ReactStreamProcessor;
 use BenTools\ETL\Recipe\Recipe;
 use BenTools\ETL\Transformer\ChainTransformer;
 use BenTools\ETL\Transformer\TransformerInterface;
@@ -76,3 +77,8 @@ function stdOut(): STDOUTLoader
 {
     return new STDOUTLoader();
 }
+
+function useReact(): EtlExecutor
+{
+    return withRecipe(new ReactStreamProcessor());
+}

tests/Behavior/ReactStreamProcessorTest.php

@@ -0,0 +1,63 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BenTools\ETL\Tests\Behavior;
+
+use BenTools\ETL\EventDispatcher\Event\ExtractEvent;
+use BenTools\ETL\Exception\ExtractException;
+use React\EventLoop\Loop;
+use React\Stream\ReadableResourceStream;
+use RuntimeException;
+
+use function BenTools\ETL\useReact;
+use function expect;
+use function fopen;
+
+it('processes React streams', function () {
+    // Given
+    $stream = new ReadableResourceStream(fopen('php://temp', 'rb'));
+    Loop::futureTick(fn () => $stream->emit('data', ['hello']));
+    Loop::futureTick(fn () => $stream->emit('data', ['world']));
+    $executor = useReact();
+
+    // When
+    $state = $executor->process($stream);
+
+    // Then
+    expect($state->output)->toBe(['hello', 'world']);
+});
+
+it('can skip items and stop the workflow', function () {
+    // Given
+    $stream = new ReadableResourceStream(fopen('php://temp', 'rb'));
+    $fruits = ['banana', 'apple', 'strawberry', 'raspberry', 'peach'];
+    foreach ($fruits as $fruit) {
+        Loop::futureTick(fn () => $stream->emit('data', [$fruit]));
+    }
+    $executor = useReact()
+        ->onExtract(function (ExtractEvent $event) {
+            match ($event->item) {
+                'apple' => $event->state->skip(),
+                'peach' => $event->state->stop(),
+                default => null,
+            };
+        })
+    ;
+
+    // When
+    $state = $executor->process($stream);
+
+    // Then
+    expect($state->output)->toBe(['banana', 'strawberry', 'raspberry']);
+});
+
+it('throws ExtractExceptions', function () {
+    // Given
+    $stream = new ReadableResourceStream(fopen('php://temp', 'rb'));
+    Loop::futureTick(fn () => $stream->emit('data', ['hello']));
+    $executor = useReact()->onExtract(fn () => throw new RuntimeException());
+
+    // When
+    $executor->process($stream);
+})->throws(ExtractException::class);

tests/Unit/EtlExecutorTest.php

@@ -9,8 +9,8 @@
 use BenTools\ETL\EtlState;
 use BenTools\ETL\EventDispatcher\Event\FlushEvent;
 use BenTools\ETL\Exception\ExtractException;
-use BenTools\ETL\Extractor\ExtractorProcessorInterface;
 use BenTools\ETL\Loader\ConditionalLoaderInterface;
+use BenTools\ETL\Processor\ProcessorInterface;
 use LogicException;
 use Pest\Exceptions\ShouldNotHappen;
 
@@ -98,7 +98,7 @@ public function flush(bool $isPartial, EtlState $state): mixed
 it('yells if it cannot process extracted data', function () {
     // Given
     $executor = (new EtlExecutor())->withProcessor(
-        new class() implements ExtractorProcessorInterface {
+        new class() implements ProcessorInterface {
             public function supports(mixed $extracted): bool
             {
                 return false;

tests/Unit/Extractor/ReactStreamExtractorTest.php

@@ -0,0 +1,21 @@
+<?php
+
+declare(strict_types=1);
+
+namespace BenTools\ETL\Tests\Unit\Extractor;
+
+use BenTools\ETL\EtlState;
+use BenTools\ETL\Extractor\ReactStreamExtractor;
+use Mockery;
+use React\Stream\ReadableStreamInterface;
+
+use function expect;
+
+it('supports any readable stream', function () {
+    $a = Mockery::mock(ReadableStreamInterface::class);
+    $b = Mockery::mock(ReadableStreamInterface::class);
+
+    $extractor = new ReactStreamExtractor($a);
+    expect($extractor->extract(new EtlState(source: $b)))->toBe($b)
+        ->and($extractor->extract(new EtlState()))->toBe($a);
+});