Obey strict method and event semantics for closed and piped streams

src/Stream.php

@@ -99,7 +99,7 @@ public function resume()
     public function write($data)
     {
         if (!$this->writable) {
-            return;
+            return false;
         }
 
         return $this->buffer->write($data);

src/ThroughStream.php

@@ -4,6 +4,8 @@
 
 class ThroughStream extends CompositeStream
 {
+    private $paused = false;
+
     public function __construct()
     {
         $readable = new ReadableStream();
@@ -17,19 +19,35 @@ public function filter($data)
         return $data;
     }
 
+    public function pause()
+    {
+        parent::pause();
+        $this->paused = true;
+    }
+
+    public function resume()
+    {
+        parent::resume();
+        $this->paused = false;
+    }
+
     public function write($data)
     {
-        if (!$this->writable) {
+        if (!$this->writable->isWritable()) {
             return false;
         }
 
         $this->readable->emit('data', array($this->filter($data)));
 
-        return true;
+        return $this->writable->isWritable() && !$this->paused;
     }
 
     public function end($data = null)
     {
+        if (!$this->writable->isWritable()) {
+            return;
+        }
+
         if (null !== $data) {
             $this->readable->emit('data', array($this->filter($data)));
         }

tests/BufferedSinkTest.php

@@ -133,6 +133,15 @@ public function writeShouldTriggerProgressOnPromise()
         $sink->end('baz');
     }
 
+    /** @test */
+    public function writeAfterEndShouldReturnFalse()
+    {
+        $sink = new BufferedSink();
+        $sink->end();
+
+        $this->assertFalse($sink->write('foo'));
+    }
+
     /** @test */
     public function forwardedErrorsFromPipeShouldRejectPromise()
     {

tests/CompositeStreamTest.php

@@ -125,6 +125,44 @@ public function itShouldHandlePipingCorrectly()
         $input->emit('data', array('foo'));
     }
 
+    /** @test */
+    public function itShouldForwardPauseUpstreamWhenPipedTo()
+    {
+        $readable = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $readable->expects($this->any())->method('isReadable')->willReturn(true);
+        $writable = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $writable->expects($this->any())->method('isWritable')->willReturn(true);
+
+        $composite = new CompositeStream($readable, $writable);
+
+        $input = $this->getMockBuilder('React\Stream\ReadableStream')->setMethods(array('pause', 'resume'))->getMock();
+        $input
+            ->expects($this->once())
+            ->method('pause');
+
+        $input->pipe($composite);
+        $composite->pause();
+    }
+
+    /** @test */
+    public function itShouldForwardResumeUpstreamWhenPipedTo()
+    {
+        $readable = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $readable->expects($this->any())->method('isReadable')->willReturn(true);
+        $writable = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $writable->expects($this->any())->method('isWritable')->willReturn(true);
+
+        $composite = new CompositeStream($readable, $writable);
+
+        $input = $this->getMockBuilder('React\Stream\ReadableStream')->setMethods(array('pause', 'resume'))->getMock();
+        $input
+            ->expects($this->once())
+            ->method('resume');
+
+        $input->pipe($composite);
+        $composite->resume();
+    }
+
     /** @test */
     public function itShouldForwardPauseAndResumeUpstreamWhenPipedTo()
     {

tests/StreamTest.php

@@ -57,6 +57,32 @@ public function testCloseShouldEmitCloseEvent()
         $this->assertFalse($conn->isReadable());
     }
 
+    public function testEndShouldEndBuffer()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+
+        $buffer = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $buffer->expects($this->once())->method('end')->with('foo');
+
+        $conn = new Stream($stream, $loop, $buffer);
+        $conn->end('foo');
+    }
+
+
+    public function testEndAfterCloseIsNoOp()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+
+        $buffer = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $buffer->expects($this->never())->method('end');
+
+        $conn = new Stream($stream, $loop);
+        $conn->close();
+        $conn->end();
+    }
+
     /**
      * @covers React\Stream\Stream::__construct
      * @covers React\Stream\Stream::handleData
@@ -193,7 +219,7 @@ public function testEndedStreamsShouldNotWrite()
         $stream = fopen($file, 'r');
 
         $this->assertSame("foo\n", fgets($stream));
-        $this->assertNull($res);
+        $this->assertFalse($res);
 
         unlink($file);
     }

tests/ThroughStreamTest.php

@@ -62,6 +62,53 @@ public function endShouldWriteDataBeforeClosing()
         $this->assertFalse($through->isWritable());
     }
 
+    /** @test */
+    public function endTwiceShouldOnlyEmitOnce()
+    {
+        $through = new ThroughStream();
+        $through->on('data', $this->expectCallableOnce('first'));
+        $through->end('first');
+        $through->end('ignored');
+    }
+
+    /** @test */
+    public function writeAfterEndShouldReturnFalse()
+    {
+        $through = new ThroughStream();
+        $through->on('data', $this->expectCallableNever());
+        $through->end();
+
+        $this->assertFalse($through->write('foo'));
+    }
+
+    /** @test */
+    public function writeDataWillCloseStreamShouldReturnFalse()
+    {
+        $through = new ThroughStream();
+        $through->on('data', array($through, 'close'));
+
+        $this->assertFalse($through->write('foo'));
+    }
+
+    /** @test */
+    public function writeDataToPausedShouldReturnFalse()
+    {
+        $through = new ThroughStream();
+        $through->pause();
+
+        $this->assertFalse($through->write('foo'));
+    }
+
+    /** @test */
+    public function writeDataToResumedShouldReturnTrue()
+    {
+        $through = new ThroughStream();
+        $through->pause();
+        $through->resume();
+
+        $this->assertTrue($through->write('foo'));
+    }
+
     /** @test */
     public function itShouldBeReadableByDefault()
     {