Add isComplete() to ConcurrentIterator (#17)

src/ConcurrentArrayIterator.php

@@ -59,6 +59,11 @@ public function getPosition(): int
         return $position;
     }
 
+    public function isComplete(): bool
+    {
+        return $this->position >= $this->size;
+    }
+
     public function dispose(): void
     {
         $this->disposed ??= new DisposedException;

src/ConcurrentChainedIterator.php

@@ -78,6 +78,11 @@ public function getPosition(): int
         return $this->iterators[$position]->getPosition();
     }
 
+    public function isComplete(): bool
+    {
+        return $this->position->get() !== null;
+    }
+
     public function dispose(): void
     {
         foreach ($this->iterators as $iterator) {

src/ConcurrentIterableIterator.php

@@ -52,6 +52,11 @@ public function getPosition(): int
         return $this->iterator->getPosition();
     }
 
+    public function isComplete(): bool
+    {
+        return $this->iterator->isComplete();
+    }
+
     public function dispose(): void
     {
         $this->iterator->dispose();

src/ConcurrentIterator.php

@@ -49,6 +49,12 @@ public function getValue(): mixed;
      */
     public function getPosition(): int;
 
+    /**
+     * @return bool {@code true} if the iterator has completed (either successfully or with an error) or {@code false}
+     * if the iterator may still emit more values.
+     */
+    public function isComplete(): bool;
+
     /**
      * Disposes the iterator, indicating the consumer is no longer interested in the iterator output.
      */

src/Internal/ConcurrentClosureIterator.php

@@ -121,6 +121,11 @@ public function getPosition(): int
         return $this->queue->getPosition();
     }
 
+    public function isComplete(): bool
+    {
+        return $this->queue->isConsumed() || $this->queue->isDisposed();
+    }
+
     public function dispose(): void
     {
         $this->queue->dispose();

src/Internal/ConcurrentFlatMapIterator.php

@@ -91,6 +91,11 @@ public function dispose(): void
         $this->iterator->dispose();
     }
 
+    public function isComplete(): bool
+    {
+        return $this->iterator->isComplete();
+    }
+
     public function getIterator(): \Traversable
     {
         while ($this->continue()) {

src/Internal/ConcurrentQueueIterator.php

@@ -40,6 +40,11 @@ public function dispose(): void
         $this->state->dispose();
     }
 
+    public function isComplete(): bool
+    {
+        return $this->state->isConsumed() || $this->state->isDisposed();
+    }
+
     public function getIterator(): \Traversable
     {
         while ($this->state->continue()) {

src/Internal/QueueState.php

@@ -166,6 +166,11 @@ public function getPosition(): int
         return $position;
     }
 
+    public function hasPending(): bool
+    {
+        return \array_key_exists($this->consumePosition, $this->emittedValues);
+    }
+
     /**
      * @see Pipeline::dispose()
      */

test/ConcurrentDelayedArrayIterator.php

@@ -64,6 +64,11 @@ public function getPosition(): int
         return $position;
     }
 
+    public function isComplete(): bool
+    {
+        return $this->position >= $this->size;
+    }
+
     public function dispose(): void
     {
         $this->disposed ??= new DisposedException;

test/QueueTest.php

@@ -39,10 +39,12 @@ public function testEmit(): void
         self::assertNull($future->await());
 
         self::assertFalse($this->queue->isComplete());
+        self::assertFalse($iterator->isComplete());
 
         $this->queue->complete();
 
         self::assertTrue($this->queue->isComplete());
+        self::assertTrue($iterator->isComplete());
 
         self::assertFalse($continue->await());
     }
@@ -54,6 +56,7 @@ public function testFail(): void
         self::assertTrue($this->queue->isComplete());
 
         $pipeline = $this->queue->pipe()->getIterator();
+        self::assertTrue($pipeline->isComplete());
 
         try {
             $pipeline->continue();
@@ -123,6 +126,8 @@ public function testEmitAfterContinue(): void
             return $pipeline->getValue();
         });
 
+        delay(0); // Enter async function above.
+
         $backPressure = $this->queue->pushAsync($value);
 
         self::assertSame($value, $future->await());
@@ -136,7 +141,11 @@ public function testEmitAfterContinue(): void
 
         self::assertNull($backPressure->await());
 
+        self::assertFalse($pipeline->isComplete());
+
         $this->queue->complete();
+
+        self::assertTrue($pipeline->isComplete());
     }
 
     public function testContinueAfterComplete(): void
@@ -210,6 +219,7 @@ public function testEmitAfterDisposal(): void
     {
         $pipeline = $this->queue->pipe();
         $pipeline->dispose();
+        self::assertTrue($pipeline->getIterator()->isComplete());
         self::assertTrue($this->queue->isDisposed());
 
         try {