CompositeStream should pause() readable side when ending

Author: clue

src/CompositeStream.php

@@ -46,6 +46,10 @@ public function pause()
 
     public function resume()
     {
+        if (!$this->writable->isWritable()) {
+            return;
+        }
+
         if ($this->pipeSource) {
             $this->pipeSource->resume();
         }
@@ -70,6 +74,7 @@ public function write($data)
 
     public function end($data = null)
     {
+        $this->readable->pause();
         $this->writable->end($data);
     }
 

tests/CompositeStreamTest.php

@@ -43,13 +43,35 @@ public function itShouldForwardReadableCallsToReadableStream()
             ->expects($this->once())
             ->method('resume');
         $writable = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $writable
+            ->expects($this->any())
+            ->method('isWritable')
+            ->willReturn(true);
 
         $composite = new CompositeStream($readable, $writable);
         $composite->isReadable();
         $composite->pause();
         $composite->resume();
     }
 
+    /** @test */
+    public function itShouldNotForwardResumeIfStreamIsNotWritable()
+    {
+        $readable = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $readable
+            ->expects($this->never())
+            ->method('resume');
+
+        $writable = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $writable
+            ->expects($this->once())
+            ->method('isWritable')
+            ->willReturn(false);
+
+        $composite = new CompositeStream($readable, $writable);
+        $composite->resume();
+    }
+
     /** @test */
     public function endShouldDelegateToWritableWithData()
     {