changes in Request, Response and tests

Author: Maciej Mroziński

composer.json

@@ -8,7 +8,7 @@
         "guzzlehttp/psr7": "^1.0",
         "react/socket": "^0.7",
         "react/event-loop": "0.4.*",
-        "react/stream": "0.4.*",
+        "react/stream": "^0.5|^0.6",
         "react/promise": "~2.2",
         "evenement/evenement": "~2.0"
     },

src/Request.php

@@ -67,6 +67,7 @@ function ($stream) use ($requestData, &$streamRef, &$stateRef) {
                     $stream->on('data', array($this, 'handleData'));
                     $stream->on('end', array($this, 'handleEnd'));
                     $stream->on('error', array($this, 'handleError'));
+                    $stream->on('close', array($this, 'handleClose'));
 
                     $headers = (string) $requestData;
 
@@ -144,14 +145,15 @@ public function handleData($data)
             $this->stream->removeListener('data', array($this, 'handleData'));
             $this->stream->removeListener('end', array($this, 'handleEnd'));
             $this->stream->removeListener('error', array($this, 'handleError'));
+            $this->stream->removeListener('close', array($this, 'handleClose'));
 
             if (!isset($response)) {
                 return;
             }
 
             $this->response = $response;
 
-            $response->on('end', function () {
+            $response->on('close', function () {
                 $this->close();
             });
             $response->on('error', function (\Exception $error) {
@@ -170,9 +172,7 @@ public function handleData($data)
 
     public function handleEnd()
     {
-        $this->closeError(new \RuntimeException(
-            "Connection closed before receiving response"
-        ));
+        $this->handleClose();
     }
 
     public function handleError($error)
@@ -184,16 +184,23 @@ public function handleError($error)
         ));
     }
 
+    public function handleClose()
+    {
+        $this->closeError(new \RuntimeException(
+            "Connection closed before receiving response"
+        ));
+    }
+
     public function closeError(\Exception $error)
     {
         if (self::STATE_END <= $this->state) {
             return;
         }
-        $this->emit('error', array($error, $this));
-        $this->close($error);
+        $this->emit('error', array($error));
+        $this->close();
     }
 
-    public function close(\Exception $error = null)
+    public function close()
     {
         if (self::STATE_END <= $this->state) {
             return;
@@ -205,7 +212,7 @@ public function close(\Exception $error = null)
             $this->stream->close();
         }
 
-        $this->emit('end', array($error, $this->response, $this));
+        $this->emit('close', array());
         $this->removeAllListeners();
     }
 

src/Response.php

@@ -2,19 +2,18 @@
 
 namespace React\HttpClient;
 
-use Evenement\EventEmitterTrait;
+use Evenement\EventEmitter;
 use React\Stream\ReadableStreamInterface;
 use React\Stream\Util;
 use React\Stream\WritableStreamInterface;
 
 /**
- * @event data ($bodyChunk, Response $thisResponse)
+ * @event data ($bodyChunk)
  * @event error
  * @event end
  */
-class Response implements ReadableStreamInterface
+class Response extends EventEmitter  implements ReadableStreamInterface
 {
-    use EventEmitterTrait;
 
     private $stream;
     private $protocol;
@@ -48,6 +47,7 @@ public function __construct(ReadableStreamInterface $stream, $protocol, $version
         $this->stream->on('data', array($this, 'handleData'));
         $this->stream->on('error', array($this, 'handleError'));
         $this->stream->on('end', array($this, 'handleEnd'));
+        $this->stream->on('close', array($this, 'handleClose'));
     }
 
     public function getProtocol()
@@ -77,34 +77,48 @@ public function getHeaders()
 
     public function handleData($data)
     {
-        $this->emit('data', array($data, $this));
+        if ($this->readable) {
+            $this->emit('data', array($data));
+        }
     }
 
     public function handleEnd()
     {
+        if (!$this->readable) {
+            return;
+        }
+        $this->emit('end', array());
         $this->close();
     }
 
     public function handleError(\Exception $error)
     {
+        if (!$this->readable) {
+            return;
+        }
         $this->emit('error', array(new \RuntimeException(
             "An error occurred in the underlying stream",
             0,
             $error
-        ), $this));
+        )));
+
+        $this->close();
+    }
 
-        $this->close($error);
+    public function handleClose()
+    {
+        $this->close();
     }
 
-    public function close(\Exception $error = null)
+    public function close()
     {
         if (!$this->readable) {
             return;
         }
 
         $this->readable = false;
 
-        $this->emit('end', array($error, $this));
+        $this->emit('close', array());
 
         $this->removeAllListeners();
         $this->stream->close();