Fix: Clear next ticks (#51)

bpolaszek · Nov 20, 2023 · 30d45fd · 30d45fd
1 parent 3335c0b
commit 30d45fd
Showing 1 changed file with 8 additions and 4 deletions.
diff --git a/src/EtlExecutor.php b/src/EtlExecutor.php
@@ -123,8 +123,8 @@ public function processItem(mixed $item, mixed $key, EtlState $state): void
     private function consumeNextTick(EtlState $state): void
     {
         foreach ($state->nextTickCallbacks as $callback) {
-            ($callback)($state);
             $state->nextTickCallbacks->detach($callback);
+            ($callback)($state);
         }
     }
 
@@ -212,10 +212,14 @@ private function flush(EtlState $state, bool $early): mixed
      */
     private function terminate(EtlState $state): EtlState
     {
-        try {
-            $this->consumeNextTick($state);
-        } catch (SkipRequest|StopRequest) {
+        // Ensure everything has been cleared
+        while (0 !== count($state->nextTickCallbacks)) {
+            try {
+                $this->consumeNextTick($state);
+            } catch (SkipRequest|StopRequest) {
+            }
         }
+
         $output = $this->flush($state->getLastVersion(), false);
 
         $state = $state->getLastVersion();