Add first() function

Author: clue

README.md

@@ -7,6 +7,7 @@ built on top of [React PHP](http://reactphp.org/).
 
 * [Usage](#usage)
   * [buffer()](#buffer)
+  * [first()](#first)
   * [unwrapReadable()](#unwrapreadable)
 * [Install](#install)
 * [License](#license)
@@ -45,6 +46,39 @@ Stream\buffer($stream)->then(function ($contents) {
 });
 ```
 
+The promise will resolve with all data chunks concatenated once the stream closes.
+
+The promise will resolve with an empty string if the stream is already closed.
+
+The promise will reject if the stream emits an error.
+
+The promise will reject if it is canceled.
+
+### first()
+
+The `first(ReadableStreamInterface $stream, $event = 'data')` function can be used to create
+a `Promise` which resolves once the given event triggers for the first time.
+
+```php
+$stream = accessSomeJsonStream();
+
+Stream\first($stream)->then(function ($chunk) {
+    echo 'The first chunk arrived: ' . $chunk;
+});
+```
+
+The promise will resolve with whatever the first event emitted or `null` if the
+event does not pass any data.
+If you do not pass a custom event name, then it will wait for the first "data"
+event and resolve with a string containing the first data chunk.
+
+The promise will reject once the stream closes – unless you're waiting for the
+"close" event, in which case it will resolve.
+
+The promise will reject if the stream is already closed.
+
+The promise will reject if it is canceled.
+
 ### unwrapReadable()
 
 The `unwrapReadable(PromiseInterface $promise)` function can be used to unwrap

src/functions.php

@@ -46,6 +46,37 @@ function buffer(ReadableStreamInterface $stream)
     });
 }
 
+/**
+ * Creates a `Promise` which resolves with the first event data
+ *
+ * @param ReadableStreamInterface $stream
+ * @param string $event
+ * @return CancellablePromiseInterface Promise<mixed, Exception>
+ */
+function first(ReadableStreamInterface $stream, $event = 'data')
+{
+    // stream already ended => reject with error
+    if (!$stream->isReadable()) {
+        return Promise\reject(new \RuntimeException('Stream already closed'));
+    }
+
+    return new Promise\Promise(function ($resolve, $reject) use ($stream, $event, &$listener) {
+        $listener = function ($data) use ($stream, $event, &$listener, $resolve) {
+            $stream->removeListener($event, $listener);
+            $resolve($data);
+        };
+        $stream->on($event, $listener);
+
+        $stream->on('close', function () use ($stream, $event, $listener, $reject) {
+            $stream->removeListener($event, $listener);
+            $reject(new \RuntimeException('Stream closed'));
+        });
+    }, function ($_, $reject) use ($stream, $event, &$listener) {
+        $stream->removeListener($event, $listener);
+        $reject(new \RuntimeException('Operation cancelled'));
+    });
+}
+
 /**
  * unwrap a `Promise` which resolves with a `ReadableStreamInterface`.
  *

tests/FirstTest.php

@@ -0,0 +1,90 @@
+<?php
+
+use React\Stream\ReadableStream;
+use Clue\React\Promise\Stream;
+use React\Promise\CancellablePromiseInterface;
+
+class FirstTest extends TestCase
+{
+    public function testClosedStreamRejects()
+    {
+        $stream = new ReadableStream();
+        $stream->close();
+
+        $promise = Stream\first($stream);
+
+        $this->expectPromiseReject($promise);
+    }
+
+    public function testPendingStreamWillNotResolve()
+    {
+        $stream = new ReadableStream();
+
+        $promise = Stream\first($stream);
+
+        $promise->then($this->expectCallableNever(), $this->expectCallableNever());
+    }
+
+    public function testClosingStreamRejects()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\first($stream);
+
+        $stream->close();
+
+        $this->expectPromiseReject($promise);
+    }
+
+    public function testClosingStreamResolvesWhenWaitingForCloseEvent()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\first($stream, 'close');
+
+        $stream->close();
+
+        $this->expectPromiseResolve($promise);
+    }
+
+    public function testEmittingDataOnStreamResolvesWithFirstEvent()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\first($stream);
+
+        $stream->emit('data', array('hello', $stream));
+        $stream->emit('data', array('world', $stream));
+        $stream->close();
+
+        $this->expectPromiseResolveWith('hello', $promise);
+    }
+
+    public function testEmittingErrorOnStreamDoesNothing()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\first($stream);
+
+        $stream->emit('error', array(new \RuntimeException('test')));
+
+        $promise->then($this->expectCallableNever(), $this->expectCallableNever());
+    }
+
+    public function testEmittingErrorResolvesWhenWaitingForErrorEvent()
+    {
+        $stream = new ReadableStream();
+        $promise = Stream\first($stream, 'error');
+
+        $stream->emit('error', array(new \RuntimeException('test')));
+
+        $this->expectPromiseResolve($promise);
+    }
+
+    public function testCancelPendingStreamWillReject()
+    {
+        $stream = new ReadableStream();
+
+        $promise = Stream\first($stream);
+
+        $promise->cancel();
+
+        $this->expectPromiseReject($promise);
+    }
+}