From c2a5be2b77b114e18d003ddfede69bb8528e95c1 Mon Sep 17 00:00:00 2001
From: Aaron Piotrowski <aaron@trowski.com>
Date: Sun, 30 Jan 2022 13:32:50 -0600
Subject: [PATCH] Cancel on disposal

---
 src/Internal/ConcurrentClosureIterator.php | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/src/Internal/ConcurrentClosureIterator.php b/src/Internal/ConcurrentClosureIterator.php
index 76ae001..65a4a28 100644
--- a/src/Internal/ConcurrentClosureIterator.php
+++ b/src/Internal/ConcurrentClosureIterator.php
@@ -34,8 +34,14 @@ public function __construct(\Closure $supplier)
         $this->supplier = $supplier;
         $this->sequence = new Sequence();
         $this->queue = new QueueState();
-        $this->sources = new \SplQueue();
+        $this->sources = $sources = new \SplQueue();
         $this->deferredCancellation = new DeferredCancellation();
+
+        $this->deferredCancellation->getCancellation()->subscribe(static function () use ($sources): void {
+            while ($sources->isEmpty()) {
+                $sources->dequeue();
+            }
+        });
     }
 
     public function continue(?Cancellation $cancellation = null): bool
@@ -117,11 +123,8 @@ public function getPosition(): int
 
     public function dispose(): void
     {
-        while (!$this->sources->isEmpty()) {
-            $this->sources->dequeue();
-        }
-
         $this->queue->dispose();
+        $this->deferredCancellation->cancel();
     }
 
     public function getIterator(): \Traversable