Automatically close response stream once client connection closes

README.md

@@ -507,6 +507,15 @@ This is just a example you could use of the streaming,
 you could also send a big amount of data via little chunks 
 or use it for body data that needs to calculated.
 
+If the request handler resolves with a response stream that is already closed,
+it will simply send an empty response body.
+If the client closes the connection while the stream is still open, the
+response stream will automatically be closed.
+If a promise is resolved with a streaming body after the client closes, the
+response stream will automatically be closed.
+The `close` event can be used to clean up any pending resources allocated
+in this case (if applicable).
+
 If the response body is a `string`, a `Content-Length` header will be added
 automatically.
 If the response body is a ReactPHP `ReadableStreamInterface` and you do not

src/ChunkedEncoder.php

@@ -52,11 +52,9 @@ public function close()
         }
 
         $this->closed = true;
-
-        $this->readable = false;
+        $this->input->close();
 
         $this->emit('close');
-
         $this->removeAllListeners();
     }
 

src/Server.php

@@ -339,15 +339,35 @@ private function handleResponseBody(ResponseInterface $response, ConnectionInter
             return $connection->end();
         }
 
-        $body = $response->getBody();
-        $stream = $body;
+        $stream = $response->getBody();
 
-        if ($response->getHeaderLine('Transfer-Encoding') === 'chunked') {
-            $stream = new ChunkedEncoder($body);
+        // close response stream if connection is already closed
+        if (!$connection->isWritable()) {
+            return $stream->close();
         }
 
         $connection->write(Psr7Implementation\str($response));
-        $stream->pipe($connection);
+
+        if ($stream->isReadable()) {
+            if ($response->getHeaderLine('Transfer-Encoding') === 'chunked') {
+                $stream = new ChunkedEncoder($stream);
+            }
+
+            // Close response stream once connection closes.
+            // Note that this TCP/IP close detection may take some time,
+            // in particular this may only fire on a later read/write attempt
+            // because we stop/pause reading from the connection once the
+            // request has been processed.
+            $connection->on('close', array($stream, 'close'));
+
+            $stream->pipe($connection);
+        } else {
+            if ($response->getHeaderLine('Transfer-Encoding') === 'chunked') {
+                $connection->write("0\r\n\r\n");
+            }
+
+            $connection->end();
+        }
     }
 
     /**

tests/FunctionalServerTest.php

@@ -16,6 +16,7 @@
 use React\Promise\Promise;
 use React\Promise\PromiseInterface;
 use React\Promise\Stream;
+use React\Stream\ThroughStream;
 
 class FunctionalServerTest extends TestCase
 {
@@ -351,6 +352,100 @@ public function testSecureHttpsOnHttpStandardPortReturnsUriWithPort()
 
         $socket->close();
     }
+
+    public function testClosedStreamFromRequestHandlerWillSendEmptyBody()
+    {
+        $loop = Factory::create();
+        $socket = new Socket(0, $loop);
+        $connector = new Connector($loop);
+
+        $stream = new ThroughStream();
+        $stream->close();
+
+        $server = new Server($socket, function (RequestInterface $request) use ($stream) {
+            return new Response(200, array(), $stream);
+        });
+
+        $result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) use ($loop) {
+            $conn->write("GET / HTTP/1.0\r\n\r\n");
+
+            return Stream\buffer($conn);
+        });
+
+        $response = Block\await($result, $loop, 1.0);
+
+        $this->assertStringStartsWith("HTTP/1.0 200 OK", $response);
+        $this->assertStringEndsWith("\r\n\r\n", $response);
+
+        $socket->close();
+    }
+
+    public function testStreamFromRequestHandlerWillBeClosedIfConnectionClosesWhileSendingBody()
+    {
+        $loop = Factory::create();
+        $socket = new Socket(0, $loop);
+        $connector = new Connector($loop);
+
+        $stream = new ThroughStream();
+        $stream->on('close', $this->expectCallableOnce());
+
+        $server = new Server($socket, function (RequestInterface $request) use ($stream) {
+            return new Response(200, array(), $stream);
+        });
+
+        $result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) use ($loop) {
+            $conn->write("GET / HTTP/1.0\r\nContent-Length: 100\r\n\r\n");
+
+            $loop->addTimer(0.1, function() use ($conn) {
+                $conn->end();
+            });
+
+            return Stream\buffer($conn);
+        });
+
+        $response = Block\await($result, $loop, 1.0);
+
+        $this->assertStringStartsWith("HTTP/1.0 200 OK", $response);
+        $this->assertStringEndsWith("\r\n\r\n", $response);
+
+        $socket->close();
+    }
+
+    public function testStreamFromRequestHandlerWillBeClosedIfConnectionClosesButWillOnlyBeDetectedOnNextWrite()
+    {
+        $loop = Factory::create();
+        $socket = new Socket(0, $loop);
+        $connector = new Connector($loop);
+
+        $stream = new ThroughStream();
+        $stream->on('close', $this->expectCallableOnce());
+
+        $server = new Server($socket, function (RequestInterface $request) use ($stream) {
+            return new Response(200, array(), $stream);
+        });
+
+        $result = $connector->connect($socket->getAddress())->then(function (ConnectionInterface $conn) use ($loop) {
+            $conn->write("GET / HTTP/1.0\r\n\r\n");
+
+            $loop->addTimer(0.1, function() use ($conn) {
+                $conn->end();
+            });
+
+            return Stream\buffer($conn);
+        });
+
+        $response = Block\await($result, $loop, 1.0);
+
+        $stream->write('nope');
+        Block\sleep(0.1, $loop);
+        $stream->write('nope');
+        Block\sleep(0.1, $loop);
+
+        $this->assertStringStartsWith("HTTP/1.0 200 OK", $response);
+        $this->assertStringEndsWith("\r\n\r\n", $response);
+
+        $socket->close();
+    }
 }
 
 function noScheme($uri)

tests/ServerTest.php

@@ -646,6 +646,166 @@ function ($data) use (&$buffer) {
         $this->assertEquals('', $buffer);
     }
 
+    public function testStreamAlreadyClosedWillSendEmptyBodyChunkedEncoded()
+    {
+        $stream = new ThroughStream();
+        $stream->close();
+
+        $server = new Server($this->socket, function (ServerRequestInterface $request) use ($stream) {
+            return new Response(200, array(), $stream);
+        });
+
+        $buffer = '';
+
+        $this->connection
+            ->expects($this->any())
+            ->method('write')
+            ->will(
+                $this->returnCallback(
+                    function ($data) use (&$buffer) {
+                        $buffer .= $data;
+                    }
+                )
+            );
+
+        $this->socket->emit('connection', array($this->connection));
+
+        $data = "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n";
+        $this->connection->emit('data', array($data));
+
+        $this->assertStringStartsWith("HTTP/1.1 200 OK\r\n", $buffer);
+        $this->assertStringEndsWith("\r\n\r\n0\r\n\r\n", $buffer);
+    }
+
+    public function testResponseStreamEndingWillSendEmptyBodyChunkedEncoded()
+    {
+        $stream = new ThroughStream();
+
+        $server = new Server($this->socket, function (ServerRequestInterface $request) use ($stream) {
+            return new Response(200, array(), $stream);
+        });
+
+        $buffer = '';
+
+        $this->connection
+            ->expects($this->any())
+            ->method('write')
+            ->will(
+                $this->returnCallback(
+                    function ($data) use (&$buffer) {
+                        $buffer .= $data;
+                    }
+                )
+            );
+
+        $this->socket->emit('connection', array($this->connection));
+
+        $data = "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n";
+        $this->connection->emit('data', array($data));
+
+        $stream->end();
+
+        $this->assertStringStartsWith("HTTP/1.1 200 OK\r\n", $buffer);
+        $this->assertStringEndsWith("\r\n\r\n0\r\n\r\n", $buffer);
+    }
+
+    public function testResponseStreamAlreadyClosedWillSendEmptyBodyPlainHttp10()
+    {
+        $stream = new ThroughStream();
+        $stream->close();
+
+        $server = new Server($this->socket, function (ServerRequestInterface $request) use ($stream) {
+            return new Response(200, array(), $stream);
+        });
+
+        $buffer = '';
+
+        $this->connection
+            ->expects($this->any())
+            ->method('write')
+            ->will(
+                $this->returnCallback(
+                    function ($data) use (&$buffer) {
+                        $buffer .= $data;
+                    }
+                )
+            );
+
+        $this->socket->emit('connection', array($this->connection));
+
+        $data = "GET / HTTP/1.0\r\nHost: localhost\r\n\r\n";
+        $this->connection->emit('data', array($data));
+
+        $this->assertStringStartsWith("HTTP/1.0 200 OK\r\n", $buffer);
+        $this->assertStringEndsWith("\r\n\r\n", $buffer);
+    }
+
+    public function testResponseStreamWillBeClosedIfConnectionIsAlreadyClosed()
+    {
+        $stream = new ThroughStream();
+        $stream->on('close', $this->expectCallableOnce());
+
+        $server = new Server($this->socket, function (ServerRequestInterface $request) use ($stream) {
+            return new Response(200, array(), $stream);
+        });
+
+        $buffer = '';
+
+        $this->connection
+            ->expects($this->any())
+            ->method('write')
+            ->will(
+                $this->returnCallback(
+                    function ($data) use (&$buffer) {
+                        $buffer .= $data;
+                    }
+                )
+            );
+
+        $this->connection = $this->getMockBuilder('React\Socket\Connection')
+            ->disableOriginalConstructor()
+            ->setMethods(
+                array(
+                    'write',
+                    'end',
+                    'close',
+                    'pause',
+                    'resume',
+                    'isReadable',
+                    'isWritable',
+                    'getRemoteAddress',
+                    'getLocalAddress',
+                    'pipe'
+                )
+            )
+            ->getMock();
+
+        $this->connection->expects($this->once())->method('isWritable')->willReturn(false);
+        $this->connection->expects($this->never())->method('write');
+        $this->connection->expects($this->never())->method('write');
+
+        $this->socket->emit('connection', array($this->connection));
+
+        $data = $this->createGetRequest();
+        $this->connection->emit('data', array($data));
+    }
+
+    public function testResponseStreamWillBeClosedIfConnectionEmitsCloseEvent()
+    {
+        $stream = new ThroughStream();
+        $stream->on('close', $this->expectCallableOnce());
+
+        $server = new Server($this->socket, function (ServerRequestInterface $request) use ($stream) {
+            return new Response(200, array(), $stream);
+        });
+
+        $this->socket->emit('connection', array($this->connection));
+
+        $data = $this->createGetRequest();
+        $this->connection->emit('data', array($data));
+        $this->connection->emit('close');
+    }
+
     public function testResponseContainsSameRequestProtocolVersionAndChunkedBodyForHttp11()
     {
         $server = new Server($this->socket, function (ServerRequestInterface $request) {