The Stdio is now a well-behaving duplex stream

Author: clue

README.md

@@ -41,8 +41,17 @@ $loop = React\EventLoop\Factory::create();
 $stdio = new Stdio($loop);
 ```
 
+See below for waiting for user input and writing output.
+Alternatively, the `Stdio` is also a well-behaving duplex stream
+(implementing React's `DuplexStreamInterface`) that emits each complete
+line as a `data` event (including the trailing newline). This is considered
+advanced usage.
+
 #### Output
 
+The `Stdio` is a well-behaving writable stream
+implementing React's `WritableStreamInterface`.
+
 The `writeln($line)` method can be used to print a line to console output.
 A trailing newline will be added automatically.
 
@@ -58,9 +67,23 @@ $stdio->write('hello');
 $stdio->write(" world\n");
 ```
 
+The `overwrite($text)` method can be used to overwrite/replace the last
+incomplete line with the given text:
+
+```php
+$stdio->write('Loading…');
+$stdio->overwrite('Done!');
+```
+
+Alternatively, you can also use the `Stdio` as a writable stream.
+You can `pipe()` any readable stream into this stream.
+
 #### Input
 
-The `Stdio` will emit a `line` event for every line read from console input.
+The `Stdio` is a well-behaving readable stream
+implementing React's `ReadableStreamInterface`.
+
+It will emit a `line` event for every line read from console input.
 The event will contain the input buffer as-is, without the trailing newline.
 You can register any number of event handlers like this:
 
@@ -79,6 +102,10 @@ Using the `line` event is the recommended way to wait for user input.
 Alternatively, using the `Readline` as a readable stream is considered advanced
 usage.
 
+Alternatively, you can also use the `Stdio` as a readable stream, which emits
+each complete line as a `data` event (including the trailing newline).
+This can be used to `pipe()` this stream into other writable streams.
+
 ### Readline
 
 The [`Readline`](#readline) class is responsible for reacting to user input and presenting a prompt to the user.

composer.json

@@ -13,7 +13,7 @@
     "require": {
         "php": ">=5.3",
         "react/event-loop": "0.3.*|0.4.*",
-        "react/stream": "0.3.*|0.4.*"
+        "react/stream": "^0.4.2"
     },
     "autoload": {
         "psr-4": { "Clue\\React\\Stdio\\": "src/" }

examples/progress.php

@@ -8,19 +8,16 @@
 $loop = React\EventLoop\Factory::create();
 
 $stdio = new Stdio($loop);
-$stdio->getInput()->close();
-
 $stdio->writeln('Will print (fake) progress and then exit');
 
 $progress = new ProgressBar($stdio);
 $progress->setMaximum(mt_rand(20, 200));
 
-$loop->addPeriodicTimer(0.2, function ($timer) use ($stdio, $progress) {
+$loop->addPeriodicTimer(0.1, function ($timer) use ($stdio, $progress) {
     $progress->advance();
 
     if ($progress->isComplete()) {
-        $stdio->overwrite();
-        $stdio->writeln("Finished processing nothing!");
+        $stdio->overwrite("Finished processing nothing!" . PHP_EOL);
 
         $stdio->end();
         $timer->cancel();

src/Stdio.php

@@ -2,17 +2,21 @@
 
 namespace Clue\React\Stdio;
 
-use React\Stream\CompositeStream;
+use Evenement\EventEmitter;
+use React\Stream\DuplexStreamInterface;
 use React\EventLoop\LoopInterface;
 use React\Stream\ReadableStreamInterface;
 use React\Stream\WritableStreamInterface;
+use React\Stream\Util;
 
-class Stdio extends CompositeStream
+class Stdio extends EventEmitter implements DuplexStreamInterface
 {
     private $input;
     private $output;
     private $readline;
 
+    private $ending = false;
+    private $closed = false;
     private $incompleteLine = '';
 
     public function __construct(LoopInterface $loop, ReadableStreamInterface $input = null, WritableStreamInterface $output = null, Readline $readline = null)
@@ -35,19 +39,27 @@ public function __construct(LoopInterface $loop, ReadableStreamInterface $input
 
         $that = $this;
 
-        // stdin emits single chars
-        $this->input->on('data', function ($data) use ($that) {
-            $that->emit('char', array($data, $that));
-        });
-
         // readline data emits a new line
         $incomplete =& $this->incompleteLine;
         $this->readline->on('data', function($line) use ($that, &$incomplete) {
             // readline emits a new line on enter, so start with a blank line
             $incomplete = '';
 
+            // emit data with trailing newline in order to preserve readable API
+            $that->emit('data', array($line . PHP_EOL));
+
+            // emit custom line event for ease of use
             $that->emit('line', array($line, $that));
         });
+
+        // handle all input events (readline forwards all input events)
+        $this->readline->on('error', array($this, 'handleError'));
+        $this->readline->on('end', array($this, 'handleEnd'));
+        $this->readline->on('close', array($this, 'handleCloseInput'));
+
+        // handle all output events
+        $this->output->on('error', array($this, 'handleError'));
+        $this->output->on('close', array($this, 'handleCloseOutput'));
     }
 
     public function pause()
@@ -60,6 +72,23 @@ public function resume()
         $this->input->resume();
     }
 
+    public function isReadable()
+    {
+        return $this->input->isReadable();
+    }
+
+    public function isWritable()
+    {
+        return $this->output->isWritable();
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
     public function handleBuffer()
     {
         $that = $this;
@@ -70,7 +99,7 @@ public function handleBuffer()
 
     public function write($data)
     {
-        if ((string)$data === '') {
+        if ($this->ending || (string)$data === '') {
             return;
         }
 
@@ -159,18 +188,33 @@ public function overwrite($data = '')
 
     public function end($data = null)
     {
+        if ($this->ending) {
+            return;
+        }
+
         if ($data !== null) {
             $this->write($data);
         }
 
+        $this->ending = true;
+
+        // clear readline output, close input and end output
         $this->readline->setInput('')->setPrompt('')->clear();
-        $this->input->pause();
+        $this->input->close();
         $this->output->end();
     }
 
     public function close()
     {
-        $this->readline->setInput('')->setPrompt('')->clear();
+        if ($this->closed) {
+            return;
+        }
+
+        $this->ending = true;
+        $this->closed = true;
+
+        // clear readline output and then close
+        $this->readline->setInput('')->setPrompt('')->clear()->close();
         $this->input->close();
         $this->output->close();
     }
@@ -194,4 +238,33 @@ private function width($str)
     {
         return mb_strwidth($str, 'utf-8') - 2 * substr_count($str, "\x08");
     }
+
+    /** @internal */
+    public function handleError(\Exception $e)
+    {
+        $this->emit('error', array($e));
+        $this->close();
+    }
+
+    /** @internal */
+    public function handleEnd()
+    {
+        $this->emit('end', array());
+    }
+
+    /** @internal */
+    public function handleCloseInput()
+    {
+        if (!$this->output->isWritable()) {
+            $this->close();
+        }
+    }
+
+    /** @internal */
+    public function handleCloseOutput()
+    {
+        if (!$this->input->isReadable()) {
+            $this->close();
+        }
+    }
 }