Implement window()

README.md

@@ -231,7 +231,7 @@ In general, Pipeline instances are mutable, meaning every Pipeline-returning met
 	// Exception: Cannot traverse an already closed generator
 	```
 
-  Although there are some cases where a pipeline can be rewound and reused just like a regular array, if you need to pause iteration and continue later, or if you need to iterate strictly once without accidental resets or exceptions, use [`$pipeline->cursor()`](#pipeline-cursor).
+  Although there are some cases where a pipeline can be rewound and reused just like a regular array, if you need to pause iteration and continue later, or if you need to iterate strictly once without accidental resets or exceptions, use [`$pipeline->cursor()`](#pipeline-cursor). If you need to rewind and replay seen elements, use [`$pipeline->window()`](#pipeline-window).
 
 - Pipeline implements `IteratorAggregate` which is not the same as `Iterator`. Where the latter needed, the pipeline can be wrapped with an `IteratorIterator`:
 
@@ -490,6 +490,46 @@ foreach ($cursor as $value) {
 $remaining = \Pipeline\take($cursor)->count();
 ```
 
+## `$pipeline->window()`
+
+Returns a rewindable iterator that caches elements for replay. Unlike `cursor()` which is forward-only, `window()` buffers seen elements allowing rewind within the buffer bounds:
+
+```php
+$pipeline = \Pipeline\fromArray([1, 2, 3, 4, 5]);
+$window = $pipeline->window();
+
+foreach ($window as $value) {
+    echo $value; // 1, 2, 3
+    if ($value === 3) {
+        break;
+    }
+}
+
+// Rewind and replay from beginning
+$window->rewind();
+foreach ($window as $value) {
+    echo $value; // 1, 2, 3, 4, 5 (full replay)
+}
+```
+
+With a size limit, oldest elements are dropped (sliding window):
+
+```php
+$window = $pipeline->window(3);  // Keep last 3 elements
+
+foreach ($window as $value) {
+    if ($value === 4) {
+        break;  // Saw: 1, 2, 3, 4
+    }
+}
+
+// Rewind - oldest element (1) was dropped
+$window->rewind();
+foreach ($window as $value) {
+    echo $value; // 2, 3, 4, 5 (buffer had [2,3,4], then continues)
+}
+```
+
 ## `$pipeline->runningVariance()`
 
 Computes online statistics for the sequence: counts, sample  mean, sample variance, standard deviation. You can access these numbers on the fly with methods such as `getCount()`, `getMean()`, `getVariance()`, `getStandardDeviation()`.

src/Helper/WindowIterator.php

@@ -0,0 +1,160 @@
+<?php
+
+/**
+ * Copyright 2017, 2018 Alexey Kopytko <alexey@kopytko.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+declare(strict_types=1);
+
+namespace Pipeline\Helper;
+
+use function count;
+
+use Countable;
+use Iterator;
+use Override;
+
+/**
+ * A rewindable iterator that caches elements for replay.
+ *
+ * Unlike CursorIterator which is forward-only, WindowIterator buffers
+ * seen elements allowing rewind within the buffer bounds.
+ *
+ * @template TKey
+ * @template TValue
+ *
+ * @implements Iterator<TKey, TValue>
+ *
+ * @final
+ */
+class WindowIterator implements Iterator, Countable
+{
+    /** @var array<int, array{TKey, TValue}> */
+    private array $buffer = [];
+
+    private int $headKey = 0;
+
+    private int $position = 0;
+
+    private bool $innerExhausted = false;
+
+    private bool $initialized = false;
+
+    /**
+     * @param Iterator<TKey, TValue> $inner
+     * @param int<1, max> $maxSize Maximum buffer size
+     */
+    public function __construct(
+        private readonly Iterator $inner,
+        private readonly int $maxSize
+    ) {}
+
+    #[Override]
+    public function current(): mixed
+    {
+        if (!$this->valid()) {
+            return null;
+        }
+
+        return $this->buffer[$this->headKey + $this->position][1];
+    }
+
+    #[Override]
+    public function key(): mixed
+    {
+        if (!$this->valid()) {
+            return null;
+        }
+
+        return $this->buffer[$this->headKey + $this->position][0];
+    }
+
+    #[Override]
+    public function next(): void
+    {
+        ++$this->position;
+
+        // If still within buffer or inner exhausted, nothing to fetch
+        if ($this->position < $this->count() || $this->innerExhausted) {
+            return;
+        }
+
+        $this->fetch();
+
+        while ($this->count() > $this->maxSize) {
+            unset($this->buffer[$this->headKey]);
+            ++$this->headKey;
+            --$this->position;
+        }
+    }
+
+    #[Override]
+    public function rewind(): void
+    {
+        $this->position = 0;
+    }
+
+    #[Override]
+    public function valid(): bool
+    {
+        $this->initialize();
+
+        return $this->position < $this->count();
+    }
+
+    private function initialize(): void
+    {
+        if ($this->initialized) {
+            return;
+        }
+        $this->initialized = true;
+
+        // If inner is already pointing at data (e.g. a started generator), capture it
+        if ($this->inner->valid()) {
+            $this->pushFromInner();
+
+            return;
+        }
+
+        $this->fetch(rewind: true);
+    }
+
+    private function fetch(bool $rewind = false): void
+    {
+        match ($rewind) {
+            false => $this->inner->next(),
+            true => $this->inner->rewind(),
+        };
+
+        if (!$this->inner->valid()) {
+            $this->innerExhausted = true;
+
+            return;
+        }
+
+        $this->pushFromInner();
+    }
+
+    private function pushFromInner(): void
+    {
+        $this->buffer[] = [$this->inner->key(), $this->inner->current()];
+    }
+
+    #[Override]
+    public function count(): int
+    {
+        return count($this->buffer);
+    }
+}

src/Standard.php

@@ -59,6 +59,7 @@
 
 use Override;
 use Pipeline\Helper\CursorIterator;
+use Pipeline\Helper\WindowIterator;
 use Traversable;
 
 /**
@@ -738,6 +739,29 @@ public function cursor(): Iterator
         return new CursorIterator($iterator);
     }
 
+    /**
+     * Returns a rewindable iterator that caches elements for replay.
+     *
+     * Unlike cursor() which is forward-only, window() buffers seen elements
+     * allowing rewind within the buffer bounds.
+     *
+     * With a size limit, oldest elements are dropped (sliding window).
+     *
+     * @param int<1, max> $size Maximum buffer size
+     * @return Iterator<TKey, TValue>
+     */
+    public function window(int $size): Iterator
+    {
+        if ($this->empty()) {
+            return new EmptyIterator();
+        }
+
+        /** @var Iterator $iterator */
+        $iterator = $this->getIterator();
+
+        return new WindowIterator($iterator, $size);
+    }
+
     /**
      * By default, returns all values regardless of keys used, discarding all keys in the process. This is a terminal operation.
      * @return list<TValue>

tests/Helper/WindowIteratorTest.php

@@ -0,0 +1,66 @@
+<?php
+
+/**
+ * Copyright 2017, 2018 Alexey Kopytko <alexey@kopytko.com>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+declare(strict_types=1);
+
+namespace Tests\Pipeline\Helper;
+
+use ArrayIterator;
+
+use function count;
+
+use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\TestCase;
+use Pipeline\Helper\WindowIterator;
+use ReflectionProperty;
+use RuntimeException;
+
+/**
+ * @internal
+ */
+#[CoversClass(WindowIterator::class)]
+final class WindowIteratorTest extends TestCase
+{
+    public function testTrimLoopTerminatesCorrectly(): void
+    {
+        $callCount = 0;
+
+        $mock = $this->getMockBuilder(WindowIterator::class)
+            ->setConstructorArgs([new ArrayIterator([1, 2, 3, 4, 5]), 3])
+            ->onlyMethods(['count'])
+            ->getMock();
+
+        $mock->method('count')
+            ->willReturnCallback(function () use (&$callCount, $mock): int {
+                if (++$callCount > 50) {
+                    throw new RuntimeException('count() called >50 times - infinite loop');
+                }
+
+                $buffer = (new ReflectionProperty(WindowIterator::class, 'buffer'))->getValue($mock);
+
+                return count($buffer);
+            });
+
+        // Consume all elements - triggers trim operations
+        foreach ($mock as $_) {
+        }
+
+        // With 5 elements and maxSize 3, count() calls should be bounded
+        $this->assertLessThan(50, $callCount);
+    }
+}