Implement waterfall()

Author: igorw

README.md

@@ -58,7 +58,6 @@ $loop->run();
 
 ## Todo
 
- * Implement waterfall()
  * Implement queue()
 
 ## Tests

src/React/Async/Util.php

@@ -71,4 +71,28 @@ public static function parallel($tasks, $callback, $errback)
             call_user_func($task, $taskCallback, $taskErrback);
         }
     }
+
+    public static function waterfall($tasks, $callback, $errback)
+    {
+        $taskCallback = function () use (&$next) {
+            call_user_func_array($next, func_get_args());
+        };
+
+        $done = function () use ($callback) {
+            call_user_func_array($callback, func_get_args());
+        };
+
+        $next = function () use (&$tasks, $taskCallback, $errback, $done) {
+            if (0 === count($tasks)) {
+                call_user_func_array($done, func_get_args());
+                return;
+            }
+
+            $task = array_shift($tasks);
+            $args = array_merge(func_get_args(), array($taskCallback, $errback));
+            call_user_func_array($task, $args);
+        };
+
+        $next();
+    }
 }

tests/React/Tests/Async/UtilWaterfallTest.php

@@ -0,0 +1,81 @@
+<?php
+
+namespace React\Tests\Async;
+
+use React\Async\Util;
+
+class UtilWaterfallTest extends TestCase
+{
+    public function testWaterfallWithoutTasks()
+    {
+        $tasks = array();
+
+        $callback = $this->createCallableMock($this->once(), array());
+        $errback = $this->createCallableMock($this->never());
+
+        Util::waterfall($tasks, $callback, $errback);
+    }
+
+    public function testWaterfallWithTasks()
+    {
+        $loop = new \React\EventLoop\StreamSelectLoop();
+
+        $tasks = array(
+            function ($callback, $errback) use ($loop) {
+                $loop->addTimer(0.05, function () use ($callback) {
+                    $callback('foo');
+                });
+            },
+            function ($foo, $callback, $errback) use ($loop) {
+                $loop->addTimer(0.05, function () use ($callback, $foo) {
+                    $callback($foo.'bar');
+                });
+            },
+            function ($bar, $callback, $errback) use ($loop) {
+                $loop->addTimer(0.05, function () use ($callback, $bar) {
+                    $callback($bar.'baz');
+                });
+            },
+        );
+
+        $callback = $this->createCallableMock($this->once(), 'foobarbaz');
+        $errback = $this->createCallableMock($this->never());
+
+        Util::waterfall($tasks, $callback, $errback);
+
+        $timer = new Timer($this);
+        $timer->start();
+
+        $loop->run();
+
+        $timer->stop();
+        $timer->assertInRange(0.15, 0.30);
+    }
+
+    public function testWaterfallWithError()
+    {
+        $called = 0;
+
+        $tasks = array(
+            function ($callback, $errback) use (&$called) {
+                $callback('foo');
+                $called++;
+            },
+            function ($foo, $callback, $errback) {
+                $e = new \RuntimeException('whoops');
+                $errback($e);
+            },
+            function ($callback, $errback) use (&$called) {
+                $callback('bar');
+                $called++;
+            },
+        );
+
+        $callback = $this->createCallableMock($this->never());
+        $errback = $this->createCallableMock($this->once());
+
+        Util::waterfall($tasks, $callback, $errback);
+
+        $this->assertSame(1, $called);
+    }
+}