The Stdio is now a well-behaving duplex stream

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,43 +2,64 @@
 
 namespace Clue\React\Stdio;
 
-use React\Stream\StreamInterface;
-use React\Stream\CompositeStream;
+use Evenement\EventEmitter;
+use React\Stream\DuplexStreamInterface;
 use React\EventLoop\LoopInterface;
-use React\Stream\ReadableStream;
-use React\Stream\Stream;
+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 $needsNewline = false;
 
-    public function __construct(LoopInterface $loop, $input = true)
+    private $ending = false;
+    private $closed = false;
+    private $incompleteLine = '';
+
+    public function __construct(LoopInterface $loop, ReadableStreamInterface $input = null, WritableStreamInterface $output = null, Readline $readline = null)
     {
-        $this->input = new Stdin($loop);
+        if ($input === null) {
+            $input = new Stdin($loop);
+        }
+
+        if ($output === null) {
+            $output = new Stdout(STDOUT);
+        }
 
-        $this->output = new Stdout(STDOUT);
+        if ($readline === null) {
+            $readline = new Readline($input, $output);
+        }
 
-        $this->readline = new Readline($this->input, $this->output);
+        $this->input = $input;
+        $this->output = $output;
+        $this->readline = $readline;
 
         $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
-        $this->readline->on('data', function($line) use ($that) {
+        $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));
         });
 
-        if (!$input) {
-            $this->pause();
-        }
+        // 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()
@@ -51,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;
@@ -61,26 +99,75 @@ public function handleBuffer()
 
     public function write($data)
     {
-        // switch back to last output position
-        $this->readline->clear();
+        if ($this->ending || (string)$data === '') {
+            return;
+        }
+
+        $out = $data;
+
+        $lastNewline = strrpos($data, "\n");
+
+        $restoreReadline = false;
+
+        if ($this->incompleteLine !== '') {
+            // the last write did not end with a newline => append to existing row
+
+            // move one line up and move cursor to last position before writing data
+            $out = "\033[A" . "\r\033[" . $this->width($this->incompleteLine) . "C" . $out;
+
+            // data contains a newline, so this will overwrite the readline prompt
+            if ($lastNewline !== false) {
+                // move cursor to beginning of readline prompt and clear line
+                // clearing is important because $data may not overwrite the whole line
+                $out = "\r\033[K" . $out;
+
+                // make sure to restore readline after this output
+                $restoreReadline = true;
+            }
+        } else {
+            // here, we're writing to a new line => overwrite readline prompt
 
-        // Erase characters from cursor to end of line
-        $this->output->write("\r\033[K");
+            // move cursor to beginning of readline prompt and clear line
+            $out = "\r\033[K" . $out;
 
-        // move one line up?
-        if ($this->needsNewline) {
-            $this->output->write("\033[A");
+            // we always overwrite the readline prompt, so restore it on next line
+            $restoreReadline = true;
         }
 
-        $this->output->write($data);
+        // following write will have have to append to this line if it does not end with a newline
+        $endsWithNewline = substr($data, -1) === "\n";
 
-        $this->needsNewline = substr($data, -1) !== "\n";
+        if ($endsWithNewline) {
+            // line ends with newline, so this is line is considered complete
+            $this->incompleteLine = '';
+        } else {
+            // always end data with newline in order to append readline on next line
+            $out .= "\n";
 
-        // repeat current prompt + linebuffer
-        if ($this->needsNewline) {
-            $this->output->write("\n");
+            if ($lastNewline === false) {
+                // contains no newline at all, everything is incomplete
+                $this->incompleteLine .= $data;
+            } else {
+                // contains a newline, everything behind it is incomplete
+                $this->incompleteLine = (string)substr($data, $lastNewline + 1);
+            }
+        }
+
+        if ($restoreReadline) {
+            // write output and restore original readline prompt and line buffer
+            $this->output->write($out);
+            $this->readline->redraw();
+        } else {
+            // restore original cursor position in readline prompt
+            $pos = $this->width($this->readline->getPrompt()) + $this->readline->getCursorCell();
+            if ($pos !== 0) {
+                // we always start at beginning of line, move right by X
+                $out .= "\033[" . $pos . "C";
+            }
+
+            // write to actual output stream
+            $this->output->write($out);
         }
-        $this->readline->redraw();
     }
 
     public function writeln($line)
@@ -90,25 +177,44 @@ public function writeln($line)
 
     public function overwrite($data = '')
     {
-        // TODO: remove existing characters
+        if ($this->incompleteLine !== '') {
+            // move one line up, move to start of line and clear everything
+            $data = "\033[A\r\033[K" . $data;
+            $this->incompleteLine = '';
+        }
 
-        $this->write("\r" . $data);
+        $this->write($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();
     }
@@ -127,4 +233,38 @@ public function getReadline()
     {
         return $this->readline;
     }
+
+    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();
+        }
+    }
 }