Merge pull request #3 from clue-labs/all

Add all() function

Author: clue

README.md

@@ -8,6 +8,7 @@ built on top of [React PHP](http://reactphp.org/).
 * [Usage](#usage)
   * [buffer()](#buffer)
   * [first()](#first)
+  * [all()](#all)
   * [unwrapReadable()](#unwrapreadable)
 * [Install](#install)
 * [License](#license)
@@ -79,6 +80,32 @@ The promise will reject if the stream is already closed.
 
 The promise will reject if it is canceled.
 
+### all()
+
+The `all(ReadableStreamInterface|WritableStreamInterface $stream, $event = 'data')`
+function can be used to create a `Promise` which resolves with an array of all the event data.
+
+```php
+$stream = accessSomeJsonStream();
+
+Stream\all($stream)->then(function ($chunks) {
+    echo 'The stream consists of ' . count($chunks) . ' chunk(s)';
+});
+```
+
+The promise will resolve with an array of whatever all events emitted or `null` if the
+events do not pass any data.
+If you do not pass a custom event name, then it will wait for all the "data"
+events and resolve with an array containing all the data chunks.
+
+The promise will resolve with an array once the stream closes.
+
+The promise will resolve with an empty array if the stream is already closed.
+
+The promise will reject if the stream emits an error.
+
+The promise will reject if it is canceled.
+
 ### unwrapReadable()
 
 The `unwrapReadable(PromiseInterface $promise)` function can be used to unwrap

src/functions.php

@@ -35,7 +35,7 @@ function buffer(ReadableStreamInterface $stream)
         $stream->on('close', function () use ($resolve, &$buffer) {
             $resolve($buffer);
         });
-    }, function ($_, $reject) use ($buffer) {
+    }, function ($_, $reject) {
         $reject(new \RuntimeException('Cancelled buffering'));
     });
 
@@ -87,6 +87,56 @@ function first(EventEmitterInterface $stream, $event = 'data')
     });
 }
 
+/**
+ * Creates a `Promise` which resolves with an array of all the event data
+ *
+ * @param ReadableStreamInterface|WritableStreamInterface $stream
+ * @param string                                          $event
+ * @return CancellablePromiseInterface Promise<string, Exception>
+ */
+function all(EventEmitterInterface $stream, $event = 'data')
+{
+    // stream already ended => resolve with empty buffer
+    if ($stream instanceof ReadableStreamInterface) {
+        // readable or duplex stream not readable => already closed
+        // a half-open duplex stream is considered closed if its readable side is closed
+        if (!$stream->isReadable()) {
+            return Promise\resolve(array());
+        }
+    } elseif ($stream instanceof WritableStreamInterface) {
+        // writable-only stream (not duplex) not writable => already closed
+        if (!$stream->isWritable()) {
+            return Promise\resolve(array());
+        }
+    }
+
+    $buffer = array();
+    $bufferer = function ($data) use (&$buffer) {
+        $buffer []= $data;
+    };
+    $stream->on($event, $bufferer);
+
+    $promise = new Promise\Promise(function ($resolve, $reject) use ($stream, &$buffer) {
+        $stream->on('error', function ($error) use ($reject) {
+            $reject(new \RuntimeException('An error occured on the underlying stream while buffering', 0, $error));
+        });
+
+        $stream->on('close', function () use ($resolve, &$buffer) {
+            $resolve($buffer);
+        });
+    }, function ($_, $reject) {
+        $reject(new \RuntimeException('Cancelled buffering'));
+    });
+
+    return $promise->then(null, function ($error) use (&$buffer, $bufferer, $stream, $event) {
+        // promise rejected => clear buffer and buffering
+        $buffer = array();
+        $stream->removeListener($event, $bufferer);
+
+        throw $error;
+    });
+}
+
 /**
  * unwrap a `Promise` which resolves with a `ReadableStreamInterface`.
  *

tests/AllTest.php

@@ -0,0 +1,102 @@
+<?php
+
+use React\Stream\ReadableStream;
+use Clue\React\Promise\Stream;
+use React\Promise\CancellablePromiseInterface;
+use React\Stream\WritableStream;
+
+class AllTest extends TestCase
+{
+    public function testClosedStreamResolvesWithEmptyBuffer()
+    {
+        $stream = new ReadableStream();
+        $stream->close();
+
+        $promise = Stream\all($stream);
+
+        $this->expectPromiseResolveWith(array(), $promise);
+    }
+
+    public function testClosedWritableStreamResolvesWithEmptyBuffer()
+    {
+        $stream = new WritableStream();
+        $stream->close();
+
+        $promise = Stream\all($stream);
+
+        $this->expectPromiseResolveWith(array(), $promise);
+    }
+
+    public function testPendingStreamWillNotResolve()
+    {
+        $stream = new ReadableStream();
+
+        $promise = Stream\all($stream);
+
+        $promise->then($this->expectCallableNever(), $this->expectCallableNever());
+    }
+
+    public function testClosingStreamResolvesWithEmptyBuffer()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\all($stream);
+
+        $stream->close();
+
+        $this->expectPromiseResolveWith(array(), $promise);
+    }
+
+    public function testClosingWritableStreamResolvesWithEmptyBuffer()
+    {
+        $stream = new WritableStream();
+        $promise = Stream\all($stream);
+
+        $stream->close();
+
+        $this->expectPromiseResolveWith(array(), $promise);
+    }
+
+    public function testEmittingDataOnStreamResolvesWithArrayOfData()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\all($stream);
+
+        $stream->emit('data', array('hello', $stream));
+        $stream->emit('data', array('world', $stream));
+        $stream->close();
+
+        $this->expectPromiseResolveWith(array('hello', 'world'), $promise);
+    }
+
+    public function testEmittingErrorOnStreamRejects()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\all($stream);
+
+        $stream->emit('error', array(new \RuntimeException('test')));
+
+        $this->expectPromiseReject($promise);
+    }
+
+    public function testEmittingErrorAfterEmittingDataOnStreamRejects()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\all($stream);
+
+        $stream->emit('data', array('hello', $stream));
+        $stream->emit('error', array(new \RuntimeException('test')));
+
+        $this->expectPromiseReject($promise);
+    }
+
+    public function testCancelPendingStreamWillReject()
+    {
+        $stream = new ReadableStream();
+
+        $promise = Stream\all($stream);
+
+        $promise->cancel();
+
+        $this->expectPromiseReject($promise);
+    }
+}