Merge pull request #2 from clue-labs/first

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|WritableStreamInterface $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

@@ -5,6 +5,8 @@
 use React\Stream\ReadableStreamInterface;
 use React\Promise;
 use React\Promise\PromiseInterface;
+use React\Stream\WritableStreamInterface;
+use Evenement\EventEmitterInterface;
 
 /**
  * Creates a `Promise` which resolves with the stream data buffer
@@ -46,6 +48,45 @@ function buffer(ReadableStreamInterface $stream)
     });
 }
 
+/**
+ * Creates a `Promise` which resolves with the first event data
+ *
+ * @param ReadableStreamInterface|WritableStreamInterface $stream
+ * @param string                                          $event
+ * @return CancellablePromiseInterface Promise<mixed, Exception>
+ */
+function first(EventEmitterInterface $stream, $event = 'data')
+{
+    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\reject(new \RuntimeException('Stream already closed'));
+        }
+    } elseif ($stream instanceof WritableStreamInterface) {
+        // writable-only stream (not duplex) not writable => already closed
+        if (!$stream->isWritable()) {
+            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,111 @@
+<?php
+
+use React\Stream\ReadableStream;
+use Clue\React\Promise\Stream;
+use React\Promise\CancellablePromiseInterface;
+use React\Stream\WritableStream;
+
+class FirstTest extends TestCase
+{
+    public function testClosedReadableStreamRejects()
+    {
+        $stream = new ReadableStream();
+        $stream->close();
+
+        $promise = Stream\first($stream);
+
+        $this->expectPromiseReject($promise);
+    }
+
+    public function testClosedWritableStreamRejects()
+    {
+        $stream = new WritableStream();
+        $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 testClosingWritableStreamRejects()
+    {
+        $stream = new WritableStream();
+        $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);
+    }
+}