Add Amp\pipe() function to pipe eventual results through a functor

CHANGELOG.md

@@ -9,6 +9,9 @@
   specifying the optional data will receive it as the third parameter
   to associated when() callbacks. If not specified callbacks are passed
   null at Argument 3.
+- New `Amp\pipe()` function allows piping of deferred values through a
+  functor when they eventually resolve. This function is the singular
+  analog of the plural `Amp\map()` function.
 
 v1.0.0-beta3
 ------------

lib/functions.php

@@ -403,6 +403,31 @@ function filter(array $promises, callable $functor) {
     return $promisor->promise();
 }
 
+/**
+ * Pipe the promised value through the specified functor once it resolves
+ *
+ * @return \Amp\Promise
+ */
+function pipe($promise, callable $functor) {
+    if (!($promise instanceof Promise)) {
+        $promise = new Success($promise);
+    }
+    $promisor = new Deferred;
+    $promise->when(function($error, $result) use ($promisor, $functor) {
+        if ($error) {
+            $promisor->fail($error);
+            return;
+        }
+        try {
+            $promisor->succeed(call_user_func($functor, $result));
+        } catch (\Exception $error) {
+            $promisor->fail($error);
+        }
+    });
+
+    return $promisor->promise();
+}
+
 /**
  * Block script execution indefinitely until the specified Promise resolves
  *

test/FunctionsTest.php

@@ -10,6 +10,37 @@ use Amp\PromiseStream;
 
 class FunctionsTest extends \PHPUnit_Framework_TestCase {
 
+    public function testPipe() {
+        $invoked = 0;
+        $promise = \Amp\pipe(21, function($r) { return $r * 2; });
+        $promise->when(function($e, $r) use (&$invoked) {
+            $invoked++;
+            $this->assertSame(42, $r);
+        });
+        $this->assertSame(1, $invoked);
+    }
+
+    public function testPipeAbortsIfOriginalPromiseFails() {
+        $invoked = 0;
+        $failure = new Failure(new \RuntimeException);
+        $promise = \Amp\pipe($failure, function(){});
+        $promise->when(function($e, $r) use (&$invoked) {
+            $invoked++;
+            $this->assertInstanceOf("RuntimeException", $e);
+        });
+        $this->assertSame(1, $invoked);
+    }
+
+    public function testPipeAbortsIfFunctorThrows() {
+        $invoked = 0;
+        $promise = \Amp\pipe(42, function(){ throw new \RuntimeException; });
+        $promise->when(function($e, $r) use (&$invoked) {
+            $invoked++;
+            $this->assertInstanceOf("RuntimeException", $e);
+        });
+        $this->assertSame(1, $invoked);
+    }
+
     public function testAllResolutionWhenNoPromiseInstancesCombined() {
         $promises = [null, 1, 2, true];
         \Amp\all($promises)->when(function($e, $r) {