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

@@ -5,8 +5,9 @@
 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";
@@ -36,11 +37,17 @@ class Readline extends EventEmitter
     private $input;
     private $output;
     private $sequencer;
+    private $closed = false;
 
     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'));
@@ -90,6 +97,9 @@ public function __construct(ReadableStreamInterface $input, WritableStreamInterf
 
         // 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'));
     }
 
     /**
@@ -570,4 +580,55 @@ private function strwidth($str)
     {
         return mb_strwidth($str, $this->encoding);
     }
+
+    /** @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->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');
+    }
 }

tests/ReadlineTest.php

@@ -483,6 +483,75 @@ 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) {