Feat: Generator fan-out in transform pipeline

src/EtlExecutor.php

@@ -134,12 +134,22 @@ private function consumeNextTick(EtlState $state): void
     private function transform(mixed $item, EtlState $state): array
     {
         try {
-            $transformResult = $this->emitTransformEvent(
-                $state,
-                TransformResult::create($this->transformer->transform($item, $state)),
-            );
+            $rawResult = TransformResult::create($this->transformer->transform($item, $state));
 
-            return [...$transformResult];
+            $allItems = [];
+            foreach ($rawResult as $singleItem) {
+                try {
+                    $singleResult = $this->emitTransformEvent(
+                        $state,
+                        TransformResult::create($singleItem),
+                    );
+                    array_push($allItems, ...$singleResult);
+                } catch (SkipRequest) {
+                    continue;
+                }
+            }
+
+            return $allItems;
         } catch (SkipRequest|StopRequest $e) {
             throw $e;
         } catch (Throwable $e) {

src/Transformer/ChainTransformer.php

@@ -5,6 +5,7 @@
 namespace BenTools\ETL\Transformer;
 
 use BenTools\ETL\EtlState;
+use Generator;
 
 final readonly class ChainTransformer implements TransformerInterface
 {
@@ -35,11 +36,30 @@ public function with(
 
     public function transform(mixed $item, EtlState $state): mixed
     {
+        $items = [$item];
+        $fanned = false;
+
         foreach ($this->transformers as $transformer) {
-            $item = $transformer->transform($item, $state);
+            $nextItems = [];
+            foreach ($items as $currentItem) {
+                $result = $transformer->transform($currentItem, $state);
+                if ($result instanceof Generator) {
+                    $fanned = true;
+                    array_push($nextItems, ...$result);
+                } else {
+                    $nextItems[] = $result;
+                }
+            }
+            $items = $nextItems;
+        }
+
+        if (!$fanned) {
+            return $items[0];
         }
 
-        return $item;
+        return (static function (array $items): Generator {
+            yield from $items;
+        })($items);
     }
 
     public static function from(TransformerInterface $transformer): self

tests/Unit/Transformer/ChainTransformerTest.php

@@ -5,77 +5,69 @@
 namespace BenTools\ETL\Tests\Unit\Transformer;
 
 use BenTools\ETL\EtlExecutor;
-use BenTools\ETL\Transformer\CallableTransformer;
 use BenTools\ETL\Transformer\ChainTransformer;
 use Generator;
 
-use function BenTools\ETL\chain;
 use function expect;
-use function implode;
 use function strrev;
 use function strtoupper;
 
-it('chains transformers', function () {
+it('chains transformers with generator fan-out', function () {
     // Given
     $input = ['foo', 'bar'];
-    $executor = new EtlExecutor(
-        transformer: new CallableTransformer(
+
+    $etl = (new EtlExecutor())
+        ->transformWith(
             fn (string $item): string => strrev($item),
-        ),
-    );
-    $executor = $executor->transformWith(
-        chain($executor->transformer)
-            ->with(function (string $item): Generator {
+            function (string $item): Generator {
                 yield $item;
                 yield strtoupper($item);
-            })
-            ->with(fn (Generator $items): array => [...$items])
-            ->with(function (array $items): array {
-                $items[] = 'hey';
-
-                return $items;
-            })
-            ->with(fn (array $items): string => implode('-', $items)),
-    );
+            },
+            fn (string $item): string => "({$item})",
+        );
 
     // When
-    $report = $executor->process($input);
+    $report = $etl->process($input);
 
     // Then
-    expect($report->output)->toBe([
-        'oof-OOF-hey',
-        'rab-RAB-hey',
-    ]);
+    expect($report->output)->toBe(['(oof)', '(OOF)', '(rab)', '(RAB)']);
 });
 
-it('silently chains transformers', function () {
+it('chains transformers without generators', function () {
     // Given
     $input = ['foo', 'bar'];
 
     $etl = (new EtlExecutor())
         ->transformWith(
             fn (string $item): string => strrev($item),
-            function (string $item): Generator {
-                yield $item;
-                yield strtoupper($item);
-            },
-            fn (Generator $items): array => [...$items],
-            function (array $items): array {
-                $items[] = 'hey';
-
-                return $items;
-            },
-            fn (array $items) => yield implode('-', $items),
+            fn (string $item): string => strtoupper($item),
         );
 
     // When
     $report = $etl->process($input);
 
     // Then
-    expect($report->output)->toBe([
-        'oof-OOF-hey',
-        'rab-RAB-hey',
-    ]);
+    expect($report->output)->toBe(['OOF', 'RAB']);
+});
+
+it('chains transformers with generator fan-out using with()', function () {
+    // Given
+    $input = ['foo', 'bar'];
+
+    $chain = (new ChainTransformer(fn (string $item): string => strrev($item)))
+        ->with(function (string $item): Generator {
+            yield $item;
+            yield strtoupper($item);
+        })
+        ->with(fn (string $item): string => "({$item})");
+
+    $etl = (new EtlExecutor())->transformWith($chain);
+
+    // When
+    $report = $etl->process($input);
+
+    // Then
+    expect($report->output)->toBe(['(oof)', '(OOF)', '(rab)', '(RAB)']);
 });
 
 it('returns self', function () {