Consistent semantics for DuplexStreamInterface::end() to ensure it SHOULD also end readable side

README.md

@@ -629,6 +629,16 @@ $stream->write('nope'); // NO-OP
 $stream->end(); // NO-OP
 ```
 
+If this stream is a `DuplexStreamInterface`, calling this method SHOULD
+also end its readable side, unless the stream supports half-open mode.
+In other words, after calling this method, these streams SHOULD switch
+into non-writable AND non-readable mode, see also `isReadable()`.
+This implies that in this case, the stream SHOULD NOT emit any `data`
+or `end` events anymore.
+Streams MAY choose to use the `pause()` method logic for this, but
+special care may have to be taken to ensure a following call to the
+`resume()` method SHOULD NOT continue emitting readable events.
+
 Note that this method should not be confused with the `close()` method.
 
 #### close()

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);
     }
 

src/DuplexResourceStream.php

@@ -146,6 +146,7 @@ public function end($data = null)
 
         $this->readable = false;
         $this->writable = false;
+        $this->pause();
 
         $this->buffer->end($data);
     }

src/WritableStreamInterface.php

@@ -262,6 +262,16 @@ public function write($data);
      * $stream->end(); // NO-OP
      * ```
      *
+     * If this stream is a `DuplexStreamInterface`, calling this method SHOULD
+     * also end its readable side, unless the stream supports half-open mode.
+     * In other words, after calling this method, these streams SHOULD switch
+     * into non-writable AND non-readable mode, see also `isReadable()`.
+     * This implies that in this case, the stream SHOULD NOT emit any `data`
+     * or `end` events anymore.
+     * Streams MAY choose to use the `pause()` method logic for this, but
+     * special care may have to be taken to ensure a following call to the
+     * `resume()` method SHOULD NOT continue emitting readable events.
+     *
      * Note that this method should not be confused with the `close()` method.
      *
      * @param mixed|string|null $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()
     {

tests/DuplexResourceStreamTest.php

@@ -237,6 +237,19 @@ public function testEnd()
         $this->assertFalse($conn->isWritable());
     }
 
+    /**
+     * @covers React\Stream\DuplexResourceStream::end
+     */
+    public function testEndRemovesReadStreamFromLoop()
+    {
+        $stream = fopen('php://temp', 'r+');
+        $loop = $this->createLoopMock();
+        $loop->expects($this->once())->method('removeReadStream');
+
+        $conn = new DuplexResourceStream($stream, $loop);
+        $conn->end('bye');
+    }
+
     public function testEndedStreamsShouldNotWrite()
     {
         $file = tempnam(sys_get_temp_dir(), 'reactphptest_');