Merge pull request #84 from clue-labs/writablestreaminterface

Add WritableResourceStream

Author: clue

README.md

@@ -35,6 +35,7 @@ descriptor based implementation with an in-memory write buffer.
     * [close()](#close-1)
   * [DuplexStreamInterface](#duplexstreaminterface)
   * [ReadableResourceStream](#readableresourcestream)
+  * [WritableResourceStream](#writableresourcestream)
 * [Usage](#usage)
 * [Install](#install)
 * [Tests](#tests)
@@ -761,12 +762,73 @@ mean it reached EOF.
 $stream->bufferSize = 8192;
 ```
 
+### WritableResourceStream
+
+The `WritableResourceStream` is a concrete implementation of the
+[`WritableStreamInterface`](#writablestreaminterface) for PHP's stream resources.
+
+This can be used to represent a write-only resource like a file stream opened in
+writable mode or a stream such as `STDOUT` or `STDERR`:
+
+```php
+$stream = new WritableResourceStream(STDOUT, $loop);
+$stream->write('hello!');
+$stream->end();
+```
+
+See also [`WritableStreamInterface`](#writablestreaminterface) for more details.
+
+The first parameter given to the constructor MUST be a valid stream resource
+that is opened for writing.
+Otherwise, it will throw an `InvalidArgumentException`:
+
+```php
+// throws InvalidArgumentException
+$stream = new WritableResourceStream(false, $loop);
+```
+
+Internally, this class tries to enable non-blocking mode on the stream resource
+which may not be supported for all stream resources.
+Most notably, this is not supported by pipes on Windows (STDOUT, STDERR etc.).
+If this fails, it will throw a `RuntimeException`:
+
+```php
+// throws RuntimeException on Windows
+$stream = new WritableResourceStream(STDOUT, $loop);
+```
+
+Once the constructor is called with a valid stream resource, this class will
+take care of the underlying stream resource.
+You SHOULD only use its public API and SHOULD NOT interfere with the underlying
+stream resource manually.
+Should you need to access the underlying stream resource, you can use the public
+`$stream` property like this:
+
+```php
+var_dump(stream_get_meta_data($stream->stream));
+```
+
+Any `write()` calls to this class will not be performaned instantly, but will
+be performaned asynchronously, once the EventLoop reports the stream resource is
+ready to accept data.
+For this, it uses an in-memory buffer string to collect all outstanding writes.
+This buffer has a soft-limit applied which defines how much data it is willing
+to accept before the caller SHOULD stop sending further data.
+It currently defaults to 64 KiB and can be controlled through the public
+`$softLimit` property like this:
+
+```php
+$stream->softLimit = 8192;
+```
+
+See also [`write()`](#write) for more details.
+
 ## Usage
 ```php
     $loop = React\EventLoop\Factory::create();
 
     $source = new React\Stream\ReadableResourceStream(fopen('omg.txt', 'r'), $loop);
-    $dest = new React\Stream\Stream(fopen('wtf.txt', 'w'), $loop);
+    $dest = new React\Stream\WritableResourceStream(fopen('wtf.txt', 'w'), $loop);
 
     $source->pipe($dest);
 

examples/benchmark-throughput.php

@@ -23,8 +23,7 @@
 
 // setup input and output streams and pipe inbetween
 $in = new React\Stream\ReadableResourceStream(fopen($if, 'r'), $loop);
-$out = new React\Stream\Stream(fopen($of, 'w'), $loop);
-$out->pause();
+$out = new React\Stream\WritableResourceStream(fopen($of, 'w'), $loop);
 $in->pipe($out);
 
 // stop input stream in $t seconds

examples/cat.php

@@ -1,16 +1,14 @@
 <?php
 
 use React\EventLoop\Factory;
-use React\Stream\Stream;
 use React\Stream\ReadableResourceStream;
+use React\Stream\WritableResourceStream;
 
 require __DIR__ . '/../vendor/autoload.php';
 
 $loop = Factory::create();
 
-$stdout = new Stream(STDOUT, $loop);
-$stdout->pause();
-
+$stdout = new WritableResourceStream(STDOUT, $loop);
 $stdin = new ReadableResourceStream(STDIN, $loop);
 $stdin->pipe($stdout);
 

src/ReadableResourceStream.php

@@ -44,7 +44,7 @@ public function __construct($stream, LoopInterface $loop)
 
         // ensure resource is opened for reading (fopen mode must contain "r" or "+")
         $meta = stream_get_meta_data($stream);
-        if (isset($meta['mode']) && strpos($meta['mode'], 'r') === strpos($meta['mode'], '+')) {
+        if (isset($meta['mode']) && $meta['mode'] !== '' && strpos($meta['mode'], 'r') === strpos($meta['mode'], '+')) {
             throw new InvalidArgumentException('Given stream resource is not opened in read mode');
         }
 

src/Stream.php

@@ -58,7 +58,7 @@ public function __construct($stream, LoopInterface $loop, WritableStreamInterfac
         }
 
         if ($buffer === null) {
-            $buffer = new Buffer($stream, $loop);
+            $buffer = new WritableResourceStream($stream, $loop);
         }
 
         $this->stream = $stream;
@@ -189,7 +189,7 @@ public function handleClose()
     }
 
     /**
-     * @return WritableStreamInterface|Buffer
+     * @return WritableStreamInterface
      */
     public function getBuffer()
     {

src/Buffer.php src/WritableResourceStream.phpsrc/Buffer.php renamed to src/WritableResourceStream.php

@@ -5,11 +5,12 @@
 use Evenement\EventEmitter;
 use React\EventLoop\LoopInterface;
 
-class Buffer extends EventEmitter implements WritableStreamInterface
+class WritableResourceStream extends EventEmitter implements WritableStreamInterface
 {
     public $stream;
-    public $listening = false;
     public $softLimit = 65536;
+
+    private $listening = false;
     private $writable = true;
     private $closed = false;
     private $loop;
@@ -21,6 +22,11 @@ public function __construct($stream, LoopInterface $loop)
             throw new \InvalidArgumentException('First parameter must be a valid stream resource');
         }
 
+        $meta = stream_get_meta_data($stream);
+        if (isset($meta['mode']) && str_replace(array('b', 't'), '', $meta['mode']) === 'r') {
+            throw new \InvalidArgumentException('Given stream resource is not opened in write mode');
+        }
+
         // this class relies on non-blocking I/O in order to not interrupt the event loop
         // e.g. pipes on Windows do not support this: https://bugs.php.net/bug.php?id=47918
         if (stream_set_blocking($stream, 0) !== true) {
@@ -87,6 +93,7 @@ public function close()
         $this->removeAllListeners();
     }
 
+    /** @internal */
     public function handleWrite()
     {
         $error = null;

tests/ReadableResourceStreamTest.php

@@ -34,6 +34,10 @@ public function testConstructorThrowsExceptionOnInvalidStream()
      */
     public function testConstructorThrowsExceptionOnWriteOnlyStream()
     {
+        if (defined('HHVM_VERSION')) {
+            $this->markTestSkipped('HHVM does not report fopen mode for STDOUT');
+        }
+
         $loop = $this->createLoopMock();
 
         $this->setExpectedException('InvalidArgumentException');

tests/StreamIntegrationTest.php

@@ -4,6 +4,7 @@
 
 use React\Stream\Stream;
 use React\EventLoop as rel;
+use React\Stream\ReadableResourceStream;
 
 class StreamIntegrationTest extends TestCase
 {
@@ -231,7 +232,7 @@ public function testReadsSingleChunkFromProcessPipe($condition, $loopFactory)
 
         $loop = $loopFactory();
 
-        $stream = new Stream(popen('echo test', 'r'), $loop);
+        $stream = new ReadableResourceStream(popen('echo test', 'r'), $loop);
         $stream->on('data', $this->expectCallableOnceWith("test\n"));
         $stream->on('end', $this->expectCallableOnce());
         $stream->on('error', $this->expectCallableNever());
@@ -250,7 +251,7 @@ public function testReadsMultipleChunksFromProcessPipe($condition, $loopFactory)
 
         $loop = $loopFactory();
 
-        $stream = new Stream(popen('echo -n a;sleep 0.1;echo -n b;sleep 0.1;echo -n c', 'r'), $loop);
+        $stream = new ReadableResourceStream(popen('echo -n a;sleep 0.1;echo -n b;sleep 0.1;echo -n c', 'r'), $loop);
 
         $buffer = '';
         $stream->on('data', function ($chunk) use (&$buffer) {
@@ -276,7 +277,7 @@ public function testReadsLongChunksFromProcessPipe($condition, $loopFactory)
 
         $loop = $loopFactory();
 
-        $stream = new Stream(popen('dd if=/dev/zero bs=12345 count=1234 2>&-', 'r'), $loop);
+        $stream = new ReadableResourceStream(popen('dd if=/dev/zero bs=12345 count=1234 2>&-', 'r'), $loop);
 
         $bytes = 0;
         $stream->on('data', function ($chunk) use (&$bytes) {
@@ -302,7 +303,7 @@ public function testReadsNothingFromProcessPipeWithNoOutput($condition, $loopFac
 
         $loop = $loopFactory();
 
-        $stream = new Stream(popen('true', 'r'), $loop);
+        $stream = new ReadableResourceStream(popen('true', 'r'), $loop);
         $stream->on('data', $this->expectCallableNever());
         $stream->on('end', $this->expectCallableOnce());
         $stream->on('error', $this->expectCallableNever());

tests/UtilTest.php

@@ -2,7 +2,7 @@
 
 namespace React\Tests\Stream;
 
-use React\Stream\Buffer;
+use React\Stream\WritableResourceStream;
 use React\Stream\ReadableStream;
 use React\Stream\Util;
 use React\Stream\WritableStream;
@@ -171,13 +171,13 @@ public function testPipeWithTooSlowWritableShouldResumeOnDrain()
         $this->assertFalse($readable->paused);
     }
 
-    public function testPipeWithBuffer()
+    public function testPipeWithWritableResourceStream()
     {
         $readable = new Stub\ReadableStreamStub();
 
         $stream = fopen('php://temp', 'r+');
         $loop = $this->createLoopMock();
-        $buffer = new Buffer($stream, $loop);
+        $buffer = new WritableResourceStream($stream, $loop);
 
         $readable->pipe($buffer);