Collector maintains limit

Author: dantleech

lib/Core/PipelineBuilder.php

@@ -66,8 +66,6 @@ public function load(array $stages): self
 
                 switch (count($stage)) {
                     case 1:
-                        if (!isset($stage[0])) {
-                        }
                         list($stage) = $stage;
                         $this->stage($stage);
                         continue 2;

lib/Extension/Core/Stage/Aggregator/CollectorAggregator.php

@@ -14,11 +14,9 @@ public function __invoke(): Generator
         list($config, $data) = yield;
 
         $rows = [];
-        $count = 0;
         while (true) {
-            if ($count++ > $config['limit']) {
-                $count = 0;
-                $rows = [];
+            if (count($rows) == $config['limit']) {
+                array_shift($rows);
             }
             $rows[] = $data;
 

tests/Unit/Extension/Core/Stage/Aggregator/CollectorAggregatorTest.php

@@ -26,4 +26,24 @@ public function testCollectsRows()
             ['one' => 'one', 'time' => 50],
         ], $result);
     }
+
+    public function testLimitsCollectionToAGivenSize()
+    {
+        $result = $this->pipeline()
+            ->stage(function () {
+                yield;
+                yield ['one' => 1];
+                yield ['one' => 2];
+                yield ['one' => 3];
+                yield ['one' => 4];
+            })
+            ->stage('aggregator/collector', ['limit' => 2])
+            ->run();
+
+        $this->assertCount(2, $result);
+        $this->assertEquals([
+            [ 'one' => 3 ],
+            [ 'one' => 4 ],
+        ], $result);
+    }
 }