Delay valve

Author: dantleech

lib/Core/Pipeline.php

@@ -30,18 +30,27 @@ public function __invoke(): Generator
             foreach ($configuredGenerators as $configuredGenerator) {
                 $generatorConfig = $configuredGenerator->config();
                 $generatorConfig = $this->replaceTokens($generatorConfig, $data);
-                $data = $configuredGenerator->generator()->send([$generatorConfig, $data]);
+                $response = $configuredGenerator->generator()->send([$generatorConfig, $data]);
 
                 if (false === $configuredGenerator->generator()->valid()) {
                     break 2;
                 }
 
-                if (false === is_array($data)) {
+                if (false === $response instanceof Signal && false === is_array($response)) {
                     throw new InvalidYieldedValue(sprintf(
-                        'All yielded values must bne arrays, got "%s"',
+                        'All yielded values must be arrays or Signals, got "%s"',
                         gettype($data)
                     ));
                 }
+
+                if ($response instanceof Signal) {
+                    switch ($response) {
+                        case Signal::continue():
+                            break 2;
+                    }
+                }
+
+                $data = $response;
             }
 
             yield $data;

lib/Core/Signal.php

@@ -0,0 +1,18 @@
+<?php
+
+namespace PhpBench\Pipeline\Core;
+
+final class Signal
+{
+    private $signal;
+
+    private function __construct(string $signal)
+    {
+        $this->signal = $signal;
+    }
+
+    public static function continue()
+    {
+        return new self('continue');
+    }
+}

lib/Extension/Core/CoreExtension.php

@@ -13,6 +13,7 @@
 use PhpBench\Pipeline\Extension\Core\Stage\Filter\KeysFilter;
 use PhpBench\Pipeline\Extension\Core\Stage\Aggregator\DescribeAggregator;
 use PhpBench\Pipeline\Extension\Core\Stage\Parameter\CounterParameter;
+use PhpBench\Pipeline\Extension\Core\Stage\Valve\DelayValve;
 
 class CoreExtension implements PipelineExtension
 {
@@ -32,6 +33,7 @@ public function __construct()
             'parameter/counter' => new CounterParameter(),
             'sampler/callable' => new CallableSampler(),
             'sampler/curl' => new CurlSampler(),
+            'valve/delay' => new DelayValve(),
             'valve/take' => new TakeValve(),
         ];
     }

lib/Extension/Core/Stage/Sampler/CurlSampler.php

@@ -32,7 +32,7 @@ public function __invoke(): Generator
                 list($config, $data) = yield $info;
             }
 
-            yield Signal::continue();
+            list($config, $data) = yield Signal::continue();
         }
 
         curl_multi_close($this->multiHandle);

lib/Extension/Core/Stage/Valve/DelayValve.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace PhpBench\Pipeline\Extension\Core\Stage\Valve;
+
+use PhpBench\Pipeline\Core\Stage;
+use Generator;
+use PhpBench\Pipeline\Core\Schema;
+
+class DelayValve implements Stage
+{
+    public function __invoke(): Generator
+    {
+        list($config, $data) = yield;
+
+        while (true) {
+            usleep($config['time']);
+            list($config, $data) = yield $data;
+        }
+    }
+
+    public function configure(Schema $schema)
+    {
+        $schema->setDefaults([
+            'time' => 10000
+        ]);
+    }
+}