Fix: Early flush precedence (#26)

bpolaszek · Nov 10, 2023 · dce646a · dce646a
1 parent b8df541
commit dce646a
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 5 deletions.
diff --git a/src/EtlState.php b/src/EtlState.php
@@ -92,10 +92,10 @@ public function getDuration(): float
     public function shouldFlush(): bool
     {
         return match (true) {
+            $this->earlyFlush => true,
             INF === $this->options->flushFrequency => false,
             0 === $this->nbLoadedItemsSinceLastFlush => false,
             0 === ($this->nbLoadedItemsSinceLastFlush % $this->options->flushFrequency) => true,
-            $this->earlyFlush => true,
             default => false,
         };
     }

diff --git a/src/Loader/CallableLoader.php b/src/Loader/CallableLoader.php
@@ -23,15 +23,19 @@ public function load(mixed $item, EtlState $state): void
         if (!is_callable($callback)) {
             throw new LoadException('Invalid destination.');
         }
-        $state->context['output'] = $callback($item, $state);
-        $state->flush();
+        $state->context[__CLASS__]['loaded'][] = $callback($item, $state);
     }
 
     /**
      * @codeCoverageIgnore
      */
     public function flush(bool $isPartial, EtlState $state): mixed
     {
-        return $state->context['output'];
+        foreach ($state->context[__CLASS__]['loaded'] ?? [] as $i => $item) {
+            $state->context[__CLASS__]['output'][] = $item;
+            unset($state->context[__CLASS__]['loaded'][$i]);
+        }
+
+        return $state->context[__CLASS__]['output'] ?? [];
     }
 }
diff --git a/tests/Unit/Loader/CallableLoaderTest.php b/tests/Unit/Loader/CallableLoaderTest.php
@@ -16,7 +16,7 @@
     $loader = new CallableLoader(function (mixed $item) use (&$items) {
         $items[] = $item;
 
-        return $items;
+        return $item;
     });
 
     // When