Also support WritableStreamInterface for first()

Author: clue

README.md

@@ -56,8 +56,8 @@ 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.
+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();

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
@@ -49,15 +51,23 @@ function buffer(ReadableStreamInterface $stream)
 /**
  * Creates a `Promise` which resolves with the first event data
  *
- * @param ReadableStreamInterface $stream
- * @param string $event
+ * @param ReadableStreamInterface|WritableStreamInterface $stream
+ * @param string                                          $event
  * @return CancellablePromiseInterface Promise<mixed, Exception>
  */
-function first(ReadableStreamInterface $stream, $event = 'data')
+function first(EventEmitterInterface $stream, $event = 'data')
 {
-    // stream already ended => reject with error
-    if (!$stream->isReadable()) {
-        return Promise\reject(new \RuntimeException('Stream already closed'));
+    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) {

tests/FirstTest.php

@@ -3,10 +3,11 @@
 use React\Stream\ReadableStream;
 use Clue\React\Promise\Stream;
 use React\Promise\CancellablePromiseInterface;
+use React\Stream\WritableStream;
 
 class FirstTest extends TestCase
 {
-    public function testClosedStreamRejects()
+    public function testClosedReadableStreamRejects()
     {
         $stream = new ReadableStream();
         $stream->close();
@@ -16,6 +17,16 @@ public function testClosedStreamRejects()
         $this->expectPromiseReject($promise);
     }
 
+    public function testClosedWritableStreamRejects()
+    {
+        $stream = new WritableStream();
+        $stream->close();
+
+        $promise = Stream\first($stream);
+
+        $this->expectPromiseReject($promise);
+    }
+
     public function testPendingStreamWillNotResolve()
     {
         $stream = new ReadableStream();
@@ -35,6 +46,16 @@ public function testClosingStreamRejects()
         $this->expectPromiseReject($promise);
     }
 
+    public function testClosingWritableStreamRejects()
+    {
+        $stream = new WritableStream();
+        $promise = Stream\first($stream);
+
+        $stream->close();
+
+        $this->expectPromiseReject($promise);
+    }
+
     public function testClosingStreamResolvesWhenWaitingForCloseEvent()
     {
         $stream = new ReadableStream();