Only close stream when the we reached end of the stream

Author: WyriHaximus

src/DuplexResourceStream.php

@@ -191,7 +191,7 @@ public function handleData($stream)
 
         if ($data !== '') {
             $this->emit('data', array($data));
-        } else{
+        } elseif (\feof($this->stream)) {
             // no data read => we reached the end and close the stream
             $this->emit('end');
             $this->close();

src/ReadableResourceStream.php

@@ -144,7 +144,7 @@ public function handleData()
 
         if ($data !== '') {
             $this->emit('data', array($data));
-        } else{
+        } elseif (\feof($this->stream)) {
             // no data read => we reached the end and close the stream
             $this->emit('end');
             $this->close();

tests/DuplexResourceStreamIntegrationTest.php

@@ -2,6 +2,7 @@
 
 namespace React\Tests\Stream;
 
+use Clue\StreamFilter as Filter;
 use React\Stream\DuplexResourceStream;
 use React\Stream\ReadableResourceStream;
 use React\EventLoop\ExtEventLoop;
@@ -342,6 +343,43 @@ public function testReadsNothingFromProcessPipeWithNoOutput($condition, $loopFac
         $loop->run();
     }
 
+    /**
+     * @covers React\Stream\ReadableResourceStream::handleData
+     * @dataProvider loopProvider
+     */
+    public function testEmptyReadShouldntFcloseStream($condition, $loopFactory)
+    {
+        if (true !== $condition()) {
+            return $this->markTestSkipped('Loop implementation not available');
+        }
+
+        $server = stream_socket_server('tcp://127.0.0.1:0');
+
+        $client = stream_socket_client(stream_socket_get_name($server, false));
+        $stream = stream_socket_accept($server);
+
+
+        // add a filter which returns an error when encountering an 'a' when reading
+        Filter\append($stream, function ($chunk) {
+            return '';
+        }, STREAM_FILTER_READ);
+
+        $loop = $loopFactory();
+
+        $conn = new DuplexResourceStream($stream, $loop);
+        $conn->on('error', $this->expectCallableNever());
+        $conn->on('data', $this->expectCallableNever());
+        $conn->on('end', $this->expectCallableNever());
+
+        fwrite($client, "foobar\n");
+
+        $conn->handleData($stream);
+
+        fclose($stream);
+        fclose($client);
+        fclose($server);
+    }
+
     private function loopTick(LoopInterface $loop)
     {
         $loop->addTimer(0, function () use ($loop) {

tests/ReadableResourceStreamTest.php

@@ -365,6 +365,25 @@ public function testDataErrorShouldEmitErrorAndClose()
         $conn->handleData($stream);
     }
 
+    /**
+     * @covers React\Stream\ReadableResourceStream::handleData
+     */
+    public function testEmptyReadShouldntFcloseStream()
+    {
+        list($stream, $_) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
+        $loop = $this->createLoopMock();
+
+        $conn = new ReadableResourceStream($stream, $loop);
+        $conn->on('error', $this->expectCallableNever());
+        $conn->on('data', $this->expectCallableNever());
+        $conn->on('end', $this->expectCallableNever());
+
+        $conn->handleData();
+
+        fclose($stream);
+        fclose($_);
+    }
+
     private function createLoopMock()
     {
         return $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();