Refactor to move response body handling to ClientRequestStream

Author: clue

src/Io/ClientRequestStream.php

@@ -4,6 +4,8 @@
 
 use Evenement\EventEmitter;
 use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
+use React\Http\Message\Response;
 use React\Promise;
 use React\Socket\ConnectionInterface;
 use React\Socket\ConnectorInterface;
@@ -172,8 +174,26 @@ public function handleData($data)
 
             $this->stream->on('close', array($this, 'handleClose'));
 
-            $this->emit('response', array($response, $this->stream));
+            assert($response instanceof ResponseInterface);
+            assert($this->stream instanceof ConnectionInterface);
+            $body = $this->stream;
+
+            // determine length of response body
+            $length = null;
+            $code = $response->getStatusCode();
+            if ($this->request->getMethod() === 'HEAD' || ($code >= 100 && $code < 200) || $code == Response::STATUS_NO_CONTENT || $code == Response::STATUS_NOT_MODIFIED) {
+                $length = 0;
+            } elseif (\strtolower($response->getHeaderLine('Transfer-Encoding')) === 'chunked') {
+                $body = new ChunkedDecoder($body);
+            } elseif ($response->hasHeader('Content-Length')) {
+                $length = (int) $response->getHeaderLine('Content-Length');
+            }
+            $response = $response->withBody($body = new ReadableBodyStream($body, $length));
+
+            // emit response with streaming response body (see `Sender`)
+            $this->emit('response', array($response, $body));
 
+            // re-emit HTTP response body to trigger body parsing if parts of it are buffered
             $this->stream->emit('data', array($bodyChunk));
         }
     }

src/Io/Sender.php

@@ -6,7 +6,6 @@
 use Psr\Http\Message\ResponseInterface;
 use React\EventLoop\LoopInterface;
 use React\Http\Client\Client as HttpClient;
-use React\Http\Message\Response;
 use React\Promise\PromiseInterface;
 use React\Promise\Deferred;
 use React\Socket\ConnectorInterface;
@@ -116,18 +115,8 @@ public function send(RequestInterface $request)
             $deferred->reject($error);
         });
 
-        $requestStream->on('response', function (ResponseInterface $response, ReadableStreamInterface $body) use ($deferred, $request) {
-            $length = null;
-            $code = $response->getStatusCode();
-            if ($request->getMethod() === 'HEAD' || ($code >= 100 && $code < 200) || $code == Response::STATUS_NO_CONTENT || $code == Response::STATUS_NOT_MODIFIED) {
-                $length = 0;
-            } elseif (\strtolower($response->getHeaderLine('Transfer-Encoding')) === 'chunked') {
-                $body = new ChunkedDecoder($body);
-            } elseif ($response->hasHeader('Content-Length')) {
-                $length = (int) $response->getHeaderLine('Content-Length');
-            }
-
-            $deferred->resolve($response->withBody(new ReadableBodyStream($body, $length)));
+        $requestStream->on('response', function (ResponseInterface $response) use ($deferred, $request) {
+            $deferred->resolve($response);
         });
 
         if ($body instanceof ReadableStreamInterface) {

tests/Io/ClientRequestStreamTest.php

@@ -35,11 +35,12 @@ public function requestShouldBindToStreamEventsAndUseconnector()
 
         $this->successfulConnectionMock();
 
-        $this->stream->expects($this->exactly(6))->method('on')->withConsecutive(
+        $this->stream->expects($this->atLeast(6))->method('on')->withConsecutive(
             array('drain', $this->identicalTo(array($request, 'handleDrain'))),
             array('data', $this->identicalTo(array($request, 'handleData'))),
             array('end', $this->identicalTo(array($request, 'handleEnd'))),
             array('error', $this->identicalTo(array($request, 'handleError'))),
+            array('close', $this->identicalTo(array($request, 'handleClose'))),
             array('close', $this->identicalTo(array($request, 'handleClose')))
         );