Fork -- probably better to do this with the existing pipeline

Author: dantleech

example/http_sampler.php

@@ -17,13 +17,14 @@
         [ 'pipeline', [ 'stages' => [
             [ 'filter/keys', [ 'keys' => [ 'url', 'total_time', 'connect_time', 'concurrency']]],
             [ 'aggregator/collector', ['limit' => 2] ],
+            //[ 'aggregator/describe', [ 'group_by' => 'url', 'describe' => 'total_time' ] ],
             [ 'encoder/json', [ 'pretty' => true ] ],
-            [ 'console/redraw' ],
+            'console/redraw',
             [ 'output/stream' ],
         ]]],
         [ 'pipeline', [ 'stages' => [
             [ 'encoder/json' ],
-            [ 'output/stream', [ 'stream' => 'report.json' ]],
+            [ 'output/stream', [ 'stream' => 'report.json', 'mode' => 'a' ]],
         ]]]
     ]]],
 ]);

lib/Extension/Core/Stage/Distribution/Fork.php

@@ -14,15 +14,20 @@ public function __invoke(): Generator
     {
         list($config, $data) = yield;
 
-        /**
-         * @var GeneratorFactory $generatorFactory */
+        /** @var GeneratorFactory $generatorFactory */
         $generatorFactory = $config['generator_factory'];
 
+        $configuredGenerators = [];
+        foreach ($config['stages'] as $stage) {
+            $configuredGenerators[] = $generatorFactory->generatorFor($stage);
+        }
+
         while (true) {
-            foreach ($config['stages'] as $stage) {
-                $generatorFactory->generatorFor($stage);
-                yield;
+            foreach ($configuredGenerators as $configuredGenerator) {
+                $configuredGenerator->generator()->send([$configuredGenerator->config(), $data]);
             }
+
+            list($config, $data) = yield $data;
         }
 
     }
@@ -33,6 +38,5 @@ public function configure(Schema $schema)
             'stages' => []
         ]);
         $schema->setRequired([ 'generator_factory' ]);
-
     }
 }

tests/Unit/Extension/Core/Stage/Distribution/ForkTest.php

@@ -11,22 +11,52 @@ public function testForksDataToAStage()
     {
         $forked = [];
         $result = $this->pipeline()
+            ->stage('valve/take', ['quantity' => 1 ])
             ->stage('distribution/fork', [
                 'stages' => [
                     function () use (&$forked) {
                         $data = yield;
-                        $forked[] = $data;
-                        $data[] = 'foobar';
-                        yield $data;
+                        $forked[] = 'Forked Data';
                     }
                 ],
             ])
-            ->generator(['Foobar'])
-            ->current();
+            ->run(['Mainline Data']);
 
         $this->assertEquals([
-            'Foobar',
+            'Mainline Data',
         ], $result, 'Main pipeline data is not affected');
 
+        $this->assertEquals([
+            'Forked Data',
+        ], $forked);
+    }
+
+    public function testForksDataToMultipleStages()
+    {
+        $forked = [];
+        $result = $this->pipeline()
+            ->stage('valve/take', ['quantity' => 1 ])
+            ->stage('distribution/fork', [
+                'stages' => [
+                    function () use (&$forked) {
+                        list($config, $data) = yield;
+                        $forked[] = 'Forked Data 1';
+                    },
+                    function () use (&$forked) {
+                        list($config, $data) = yield;
+                        $forked[] = 'Forked Data 2';
+                    }
+                ],
+            ])
+            ->run(['Mainline Data']);
+
+        $this->assertEquals([
+            'Mainline Data',
+        ], $result, 'Main pipeline data is not affected');
+
+        $this->assertEquals([
+            'Forked Data 1',
+            'Forked Data 2',
+        ], $forked);
     }
 }