The Readline is now a well behaving readable stream

README.md

@@ -75,6 +75,10 @@ $stdio->on('line', function ($line) {
 You can control various aspects of the console input through the [`Readline`](#readline),
 so read on..
 
+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.
+
 ### Readline
 
 The [`Readline`](#readline) class is responsible for reacting to user input and presenting a prompt to the user.
@@ -89,6 +93,12 @@ You can access the current instance through the [`Stdio`](#stdio):
 $readline = $stdio->getReadline();
 ```
 
+See above for waiting for user input.
+Alternatively, the `Readline` is also a well-behaving readable stream
+(implementing React's `ReadableStreamInterface`) that emits each complete
+line as a `data` event (without the trailing newline). This is considered
+advanced usage.
+
 #### Prompt
 
 The *prompt* will be written at the beginning of the *user input line*, right before the *user input buffer*.

src/Readline.php

@@ -3,8 +3,11 @@
 namespace Clue\React\Stdio;
 
 use Evenement\EventEmitter;
+use React\Stream\ReadableStreamInterface;
+use React\Stream\WritableStreamInterface;
+use React\Stream\Util;
 
-class Readline extends EventEmitter
+class Readline extends EventEmitter implements ReadableStreamInterface
 {
     const KEY_BACKSPACE = "\x7f";
     const KEY_ENTER = "\n";
@@ -31,13 +34,20 @@ class Readline extends EventEmitter
     private $history = null;
     private $encoding = 'utf-8';
 
+    private $input;
     private $output;
     private $sequencer;
+    private $closed = false;
 
-    public function __construct($output)
+    public function __construct(ReadableStreamInterface $input, WritableStreamInterface $output)
     {
+        $this->input = $input;
         $this->output = $output;
 
+        if (!$this->input->isReadable()) {
+            return $this->close();
+        }
+
         $this->sequencer = new Sequencer();
         $this->sequencer->addSequence(self::KEY_ENTER, array($this, 'onKeyEnter'));
         $this->sequencer->addSequence(self::KEY_BACKSPACE, array($this, 'onKeyBackspace'));
@@ -84,6 +94,12 @@ public function __construct($output)
         $this->sequencer->addFallback(self::ESC_SEQUENCE, function ($bytes) {
             echo 'unknown sequence: ' . ord($bytes) . PHP_EOL;
         });
+
+        // input data emits a single char into readline
+        $input->on('data', array($this->sequencer, 'push'));
+        $input->on('end', array($this, 'handleEnd'));
+        $input->on('error', array($this, 'handleError'));
+        $input->on('close', array($this, 'close'));
     }
 
     /**
@@ -369,7 +385,7 @@ public function redraw()
             // write output, then move back $reverse chars (by sending backspace)
             $output .= $buffer . str_repeat("\x08", $this->strwidth($buffer) - $this->getCursorCell());
         }
-        $this->write($output);
+        $this->output->write($output);
 
         return $this;
     }
@@ -389,18 +405,12 @@ public function redraw()
     public function clear()
     {
         if ($this->prompt !== '' || ($this->echo !== false && $this->linebuffer !== '')) {
-            $this->write("\r\033[K");
+            $this->output->write("\r\033[K");
         }
 
         return $this;
     }
 
-    /** @internal */
-    public function onChar($char)
-    {
-        $this->sequencer->push($char);
-    }
-
     /** @internal */
     public function onKeyBackspace()
     {
@@ -449,7 +459,7 @@ public function onKeyTab()
     public function onKeyEnter()
     {
         if ($this->echo !== false) {
-            $this->write("\n");
+            $this->output->write("\n");
         }
         $this->processLine();
     }
@@ -571,8 +581,54 @@ private function strwidth($str)
         return mb_strwidth($str, $this->encoding);
     }
 
-    protected function write($data)
+    /** @internal */
+    public function handleEnd()
+    {
+        if (!$this->closed) {
+            $this->emit('end');
+            $this->close();
+        }
+    }
+
+    /** @internal */
+    public function handleError(\Exception $error)
+    {
+        $this->emit('error', array($error));
+        $this->close();
+    }
+
+    public function isReadable()
+    {
+        return !$this->closed && $this->input->isReadable();
+    }
+
+    public function pause()
+    {
+        $this->input->pause();
+    }
+
+    public function resume()
     {
-        $this->output->write($data);
+        $this->input->resume();
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        $this->closed = true;
+
+        $this->input->close();
+
+        $this->emit('close');
     }
 }

src/Stdio.php

@@ -22,18 +22,17 @@ public function __construct(LoopInterface $loop, $input = true)
 
         $this->output = new Stdout(STDOUT);
 
-        $this->readline = $readline = new Readline($this->output);
+        $this->readline = new Readline($this->input, $this->output);
 
         $that = $this;
 
-        // input data emits a single char into readline
-        $this->input->on('data', function ($data) use ($that, $readline) {
+        // stdin emits single chars
+        $this->input->on('data', function ($data) use ($that) {
             $that->emit('char', array($data, $that));
-            $readline->onChar($data);
         });
 
         // readline data emits a new line
-        $readline->on('data', function($line) use ($that) {
+        $this->readline->on('data', function($line) use ($that) {
             $that->emit('line', array($line, $that));
         });
 

tests/ReadlineTest.php

@@ -1,13 +1,20 @@
 <?php
 
 use Clue\React\Stdio\Readline;
+use React\Stream\ReadableStream;
 
 class ReadlineTest extends TestCase
 {
+    private $input;
+    private $output;
+    private $readline;
+
     public function setUp()
     {
-        $this->output = $this->getMockBuilder('Clue\React\Stdio\Stdout')->disableOriginalConstructor()->getMock();
-        $this->readline = new Readline($this->output);
+        $this->input = new ReadableStream();
+        $this->output = $this->getMock('React\Stream\WritableStreamInterface');
+
+        $this->readline = new Readline($this->input, $this->output);
     }
 
     public function testSettersReturnSelf()
@@ -476,10 +483,79 @@ public function testSetInputDuringEmitKeepsInput()
         $this->assertEquals('test', $readline->getInput());
     }
 
+    public function testEmitErrorWillEmitErrorAndClose()
+    {
+        $this->readline->on('error', $this->expectCallableOnce());
+        $this->readline->on('close', $this->expectCallableOnce());
+
+        $this->input->emit('error', array(new \RuntimeException()));
+
+        $this->assertFalse($this->readline->isReadable());
+    }
+
+    public function testEmitEndWillEmitEndAndClose()
+    {
+        $this->readline->on('end', $this->expectCallableOnce());
+        $this->readline->on('close', $this->expectCallableOnce());
+
+        $this->input->emit('end');
+
+        $this->assertFalse($this->readline->isReadable());
+    }
+
+    public function testEmitCloseWillEmitClose()
+    {
+        $this->readline->on('end', $this->expectCallableNever());
+        $this->readline->on('close', $this->expectCallableOnce());
+
+        $this->input->emit('close');
+
+        $this->assertFalse($this->readline->isReadable());
+    }
+
+    public function testClosedStdinWillCloseReadline()
+    {
+        $this->input = $this->getMock('React\Stream\ReadableStreamInterface');
+        $this->input->expects($this->once())->method('isReadable')->willReturn(false);
+
+        $this->readline = new Readline($this->input, $this->output);
+
+        $this->assertFalse($this->readline->isReadable());
+    }
+
+    public function testPauseWillBeForwarded()
+    {
+        $this->input = $this->getMock('React\Stream\ReadableStreamInterface');
+        $this->input->expects($this->once())->method('pause');
+
+        $this->readline = new Readline($this->input, $this->output);
+
+        $this->readline->pause();
+    }
+
+    public function testResumeWillBeForwarded()
+    {
+        $this->input = $this->getMock('React\Stream\ReadableStreamInterface');
+        $this->input->expects($this->once())->method('resume');
+
+        $this->readline = new Readline($this->input, $this->output);
+
+        $this->readline->resume();
+    }
+
+    public function testPipeWillReturnDest()
+    {
+        $dest = $this->getMock('React\Stream\WritableStreamInterface');
+
+        $ret = $this->readline->pipe($dest);
+
+        $this->assertEquals($dest, $ret);
+    }
+
     private function pushInputBytes(Readline $readline, $bytes)
     {
         foreach (str_split($bytes, 1) as $byte) {
-            $readline->onChar($byte);
+            $this->input->emit('data', array($byte));
         }
     }
 }

tests/StdioTest.php

@@ -0,0 +1,19 @@
+<?php
+
+use React\EventLoop\Factory;
+use Clue\React\Stdio\Stdio;
+
+class StdioTest extends TestCase
+{
+    private $loop;
+
+    public function setUp()
+    {
+        $this->loop = Factory::create();
+    }
+
+    public function testCtor()
+    {
+        $stdio = new Stdio($this->loop);
+    }
+}