Explicitly close connection after response body ends

src/Message/MessageFactory.php

@@ -41,7 +41,23 @@ public function request($method, $uri, $headers = array(), $content = '')
      */
     public function response($version, $status, $reason, $headers = array(), $body = '')
     {
-        return new Response($status, $headers, $this->body($body), $version, $reason);
+        $response = new Response($status, $headers, $body instanceof ReadableStreamInterface ? null : $body, $version, $reason);
+
+        if ($body instanceof ReadableStreamInterface) {
+            $length = null;
+            $code = $response->getStatusCode();
+            if (($code >= 100 && $code < 200) || $code == 204 || $code == 304) {
+                $length = 0;
+            } elseif (\strtolower($response->getHeaderLine('Transfer-Encoding')) === 'chunked') {
+                $length = null;
+            } elseif ($response->hasHeader('Content-Length')) {
+                $length = (int)$response->getHeaderLine('Content-Length');
+            }
+
+            $response = $response->withBody(new ReadableBodyStream($body, $length));
+        }
+
+        return $response;
     }
 
     /**

src/Message/ReadableBodyStream.php

@@ -14,24 +14,30 @@
 class ReadableBodyStream extends EventEmitter implements ReadableStreamInterface, StreamInterface
 {
     private $input;
+    private $position = 0;
+    private $size;
     private $closed = false;
 
-    public function __construct(ReadableStreamInterface $input)
+    public function __construct(ReadableStreamInterface $input, $size = null)
     {
         $this->input = $input;
+        $this->size = $size;
 
         $that = $this;
-        $input->on('data', function ($data) use ($that) {
+        $pos =& $this->position;
+        $input->on('data', function ($data) use ($that, &$pos, $size) {
             $that->emit('data', array($data));
+
+            $pos += \strlen($data);
+            if ($size !== null && $pos >= $size) {
+                $that->handleEnd();
+            }
         });
         $input->on('error', function ($error) use ($that) {
             $that->emit('error', array($error));
             $that->close();
         });
-        $input->on('end', function () use ($that) {
-            $that->emit('end');
-            $that->close();
-        });
+        $input->on('end', array($that, 'handleEnd'));
         $input->on('close', array($that, 'close'));
     }
 
@@ -85,7 +91,7 @@ public function detach()
 
     public function getSize()
     {
-        return null;
+        return $this->size;
     }
 
     public function tell()
@@ -132,4 +138,16 @@ public function getMetadata($key = null)
     {
         return ($key === null) ? array() : null;
     }
+
+    /** @internal */
+    public function handleEnd()
+    {
+        if ($this->position !== $this->size && $this->size !== null) {
+            $this->emit('error', array(new \UnderflowException('Unexpected end of response body after ' . $this->position . '/' . $this->size . ' bytes')));
+        } else {
+            $this->emit('end');
+        }
+
+        $this->close();
+    }
 }

tests/FunctionalBrowserTest.php

@@ -288,10 +288,101 @@ public function testPostString()
         $this->assertEquals('hello world', $data['data']);
     }
 
+    public function testReceiveStreamUntilConnectionsEndsForHttp10()
+    {
+        $loop = $this->loop;
+        $server = new StreamingServer(function (ServerRequestInterface $request) use ($loop) {
+            $stream = new ThroughStream();
+
+            $loop->futureTick(function () use ($stream) {
+                $stream->end('hello');
+            });
+
+            return new Response(
+                200,
+                array(),
+                $stream
+            );
+        });
+
+        $socket = new \React\Socket\Server(0, $this->loop);
+        $server->listen($socket);
+
+        $this->base = str_replace('tcp:', 'http:', $socket->getAddress()) . '/';
+
+        $response = Block\await($this->browser->get($this->base . 'get', array()), $this->loop);
+
+        $this->assertEquals('1.0', $response->getProtocolVersion());
+        $this->assertFalse($response->hasHeader('Transfer-Encoding'));
+        $this->assertEquals('hello', (string)$response->getBody());
+
+        $socket->close();
+    }
+
+    public function testReceiveStreamChunkedForHttp11()
+    {
+        $loop = $this->loop;
+        $server = new StreamingServer(function (ServerRequestInterface $request) use ($loop) {
+            $stream = new ThroughStream();
+
+            $loop->futureTick(function () use ($stream) {
+                $stream->end('hello');
+            });
+
+            return new Response(
+                200,
+                array(),
+                $stream
+            );
+        });
+
+        $socket = new \React\Socket\Server(0, $this->loop);
+        $server->listen($socket);
+
+        $this->base = str_replace('tcp:', 'http:', $socket->getAddress()) . '/';
+
+        $response = Block\await($this->browser->send(new Request('GET', $this->base . 'get', array(), null, '1.1')), $this->loop);
+
+        $this->assertEquals('1.1', $response->getProtocolVersion());
+
+        // underlying http-client automatically decodes and doesn't expose header
+        // @link https://github.com/reactphp/http-client/pull/58
+        // $this->assertEquals('chunked', $response->getHeaderLine('Transfer-Encoding'));
+        $this->assertFalse($response->hasHeader('Transfer-Encoding'));
+
+        $this->assertEquals('hello', (string)$response->getBody());
+
+        $socket->close();
+    }
+
+    public function testReceiveStreamAndExplicitlyCloseConnectionEvenWhenServerKeepsConnectionOpen()
+    {
+        $closed = new \React\Promise\Deferred();
+        $socket = new \React\Socket\Server(0, $this->loop);
+        $socket->on('connection', function (\React\Socket\ConnectionInterface $connection) use ($closed) {
+            $connection->on('data', function () use ($connection) {
+                $connection->write("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello");
+            });
+            $connection->on('close', function () use ($closed) {
+                $closed->resolve(true);
+            });
+        });
+
+        $this->base = str_replace('tcp:', 'http:', $socket->getAddress()) . '/';
+
+        $response = Block\await($this->browser->get($this->base . 'get', array()), $this->loop);
+        $this->assertEquals('hello', (string)$response->getBody());
+
+        $ret = Block\await($closed->promise(), $this->loop, 0.1);
+        $this->assertTrue($ret);
+
+        $socket->close();
+    }
+
     public function testPostStreamChunked()
     {
         // httpbin used to support `Transfer-Encoding: chunked` for requests,
-        // but not rejects those, so let's start our own server instance
+        // but now rejects these, so let's start our own server instance
         $that = $this;
         $server = new StreamingServer(function (ServerRequestInterface $request) use ($that) {
             $that->assertFalse($request->hasHeader('Content-Length'));

tests/Message/MessageFactoryTest.php

@@ -96,4 +96,87 @@ public function testBodyReadableStream()
         $this->assertEquals(null, $body->getSize());
         $this->assertEquals('', (string)$body);
     }
+
+    public function testResponseWithBodyString()
+    {
+        $response = $this->messageFactory->response('1.1', 200, 'OK', array(), 'hi');
+
+        $body = $response->getBody();
+        $this->assertInstanceOf('Psr\Http\Message\StreamInterface', $body);
+        $this->assertNotInstanceOf('React\Stream\ReadableStreamInterface', $body);
+        $this->assertEquals(2, $body->getSize());
+        $this->assertEquals('hi', (string)$body);
+    }
+
+    public function testResponseWithStreamingBodyHasUnknownSizeByDefault()
+    {
+        $stream = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $response = $this->messageFactory->response('1.1', 200, 'OK', array(), $stream);
+
+        $body = $response->getBody();
+        $this->assertInstanceOf('Psr\Http\Message\StreamInterface', $body);
+        $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $body);
+        $this->assertNull($body->getSize());
+        $this->assertEquals('', (string)$body);
+    }
+
+    public function testResponseWithStreamingBodyHasSizeFromContentLengthHeader()
+    {
+        $stream = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $response = $this->messageFactory->response('1.1', 200, 'OK', array('Content-Length' => '100'), $stream);
+
+        $body = $response->getBody();
+        $this->assertInstanceOf('Psr\Http\Message\StreamInterface', $body);
+        $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $body);
+        $this->assertEquals(100, $body->getSize());
+        $this->assertEquals('', (string)$body);
+    }
+
+    public function testResponseWithStreamingBodyHasUnknownSizeWithTransferEncodingChunkedHeader()
+    {
+        $stream = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $response = $this->messageFactory->response('1.1', 200, 'OK', array('Transfer-Encoding' => 'chunked'), $stream);
+
+        $body = $response->getBody();
+        $this->assertInstanceOf('Psr\Http\Message\StreamInterface', $body);
+        $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $body);
+        $this->assertNull($body->getSize());
+        $this->assertEquals('', (string)$body);
+    }
+
+    public function testResponseWithStreamingBodyHasZeroSizeForInformationalResponse()
+    {
+        $stream = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $response = $this->messageFactory->response('1.1', 101, 'OK', array('Content-Length' => '100'), $stream);
+
+        $body = $response->getBody();
+        $this->assertInstanceOf('Psr\Http\Message\StreamInterface', $body);
+        $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $body);
+        $this->assertEquals(0, $body->getSize());
+        $this->assertEquals('', (string)$body);
+    }
+
+    public function testResponseWithStreamingBodyHasZeroSizeForNoContentResponse()
+    {
+        $stream = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $response = $this->messageFactory->response('1.1', 204, 'OK', array('Content-Length' => '100'), $stream);
+
+        $body = $response->getBody();
+        $this->assertInstanceOf('Psr\Http\Message\StreamInterface', $body);
+        $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $body);
+        $this->assertEquals(0, $body->getSize());
+        $this->assertEquals('', (string)$body);
+    }
+
+    public function testResponseWithStreamingBodyHasZeroSizeForNotModifiedResponse()
+    {
+        $stream = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
+        $response = $this->messageFactory->response('1.1', 304, 'OK', array('Content-Length' => '100'), $stream);
+
+        $body = $response->getBody();
+        $this->assertInstanceOf('Psr\Http\Message\StreamInterface', $body);
+        $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $body);
+        $this->assertEquals(0, $body->getSize());
+        $this->assertEquals('', (string)$body);
+    }
 }

tests/Message/ReadableBodyStreamTest.php

@@ -86,6 +86,84 @@ public function testEndInputWillEmitCloseEvent()
         $this->assertEquals(1, $called);
     }
 
+    public function testEndInputWillEmitErrorEventWhenDataDoesNotReachExpectedLength()
+    {
+        $this->input = new ThroughStream();
+        $this->stream = new ReadableBodyStream($this->input, 5);
+
+        $called = null;
+        $this->stream->on('error', function ($e) use (&$called) {
+            $called = $e;
+        });
+
+        $this->input->write('hi');
+        $this->input->end();
+
+        $this->assertInstanceOf('UnderflowException', $called);
+        $this->assertSame('Unexpected end of response body after 2/5 bytes', $called->getMessage());
+    }
+
+    public function testDataEventOnInputWillEmitDataEvent()
+    {
+        $this->input = new ThroughStream();
+        $this->stream = new ReadableBodyStream($this->input);
+
+        $called = null;
+        $this->stream->on('data', function ($data) use (&$called) {
+            $called = $data;
+        });
+
+        $this->input->write('hello');
+
+        $this->assertEquals('hello', $called);
+    }
+
+    public function testDataEventOnInputWillEmitEndWhenDataReachesExpectedLength()
+    {
+        $this->input = new ThroughStream();
+        $this->stream = new ReadableBodyStream($this->input, 5);
+
+        $called = null;
+        $this->stream->on('end', function () use (&$called) {
+            ++$called;
+        });
+
+        $this->input->write('hello');
+
+        $this->assertEquals(1, $called);
+    }
+
+    public function testEndEventOnInputWillEmitEndOnlyOnceWhenDataAlreadyReachedExpectedLength()
+    {
+        $this->input = new ThroughStream();
+        $this->stream = new ReadableBodyStream($this->input, 5);
+
+        $called = null;
+        $this->stream->on('end', function () use (&$called) {
+            ++$called;
+        });
+
+        $this->input->write('hello');
+        $this->input->end();
+
+        $this->assertEquals(1, $called);
+    }
+
+    public function testDataEventOnInputWillNotEmitEndWhenDataDoesNotReachExpectedLength()
+    {
+        $this->input = new ThroughStream();
+        $this->stream = new ReadableBodyStream($this->input, 5);
+
+        $called = null;
+        $this->stream->on('end', function () use (&$called) {
+            ++$called;
+        });
+
+        $this->input->write('hi');
+
+        $this->assertNull($called);
+    }
+
     public function testPauseWillPauseInputStream()
     {
         $this->input->expects($this->once())->method('pause');