Fix clearing waiting and back-pressure on completion

Fixes #22.
amphp · Jul 4, 2024 · 66c0956 · 66c0956
1 parent f1c2ce3
commit 66c0956
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 11 deletions.
diff --git a/src/Internal/QueueState.php b/src/Internal/QueueState.php
@@ -386,7 +386,7 @@ private function finalize(?\Throwable $exception = null, bool $disposed = false)
     private function relieveBackPressure(?\Throwable $exception): void
     {
         $backPressure = $this->backpressure;
-        unset($this->backpressure);
+        $this->backpressure = [];
 
         foreach ($backPressure as $placeholder) {
             if ($exception) {
@@ -407,17 +407,13 @@ private function relieveBackPressure(?\Throwable $exception): void
     private function resolvePending(): void
     {
         $waiting = $this->waiting;
-        unset($this->waiting);
+        $this->waiting = [];
 
-        $exception = $this->exception ?? null;
-
-        if ($waiting) {
-            foreach ($waiting as $suspension) {
-                if ($exception) {
-                    $suspension->throw($exception);
-                } else {
-                    $suspension->resume($this->currentPosition);
-                }
+        foreach ($waiting as $suspension) {
+            if ($this->exception) {
+                $suspension->throw($this->exception);
+            } else {
+                $suspension->resume($this->currentPosition);
             }
         }
     }

diff --git a/test/QueueTest.php b/test/QueueTest.php
@@ -566,4 +566,23 @@ public function provideBufferSize(): iterable
         yield 'buffer size 5' => [5];
         yield 'buffer size 10' => [10];
     }
+
+    public function testCompleteWhileContinueCancellationPending(): void
+    {
+        $queue = new Queue();
+        $iterator = $queue->iterate();
+
+        $deferredCancellation = new DeferredCancellation();
+
+        $future1 = async(fn () => $iterator->continue($deferredCancellation->getCancellation()));
+
+        $future2 = async(function () use ($queue, $deferredCancellation): void {
+            $deferredCancellation->cancel();
+            $queue->complete();
+        });
+
+        $future2->await();
+
+        self::assertFalse($future1->await());
+    }
 }