Writable updated

src/Io/ClientRequestStream.php

@@ -183,7 +183,9 @@ public function handleData($data)
                 if ($successfulEndReceived && $connection->isReadable() && $that->hasMessageKeepAliveEnabled($response) && $that->hasMessageKeepAliveEnabled($request)) {
                     $connectionManager->keepAlive($request->getUri(), $connection);
                 } else {
-                    $connection->close();
+                    if (! $that->responseIsAnUpgradeResponse($response)) {
+                        $connection->close();
+                    }
                 }
 
                 $that->close();
@@ -199,7 +201,12 @@ public function handleData($data)
             } elseif ($response->hasHeader('Content-Length')) {
                 $length = (int) $response->getHeaderLine('Content-Length');
             }
-            $response = $response->withBody($body = new ReadableBodyStream($body, $length));
+
+            $body = $this->responseIsAnUpgradeResponse($response)
+                ? new DuplexBodyStream($connection)
+                : new ReadableBodyStream($body, $length);
+
+            $response = $response->withBody($body);
             $body->on('end', function () use (&$successfulEndReceived) {
                 $successfulEndReceived = true;
             });
@@ -216,6 +223,14 @@ public function handleData($data)
         }
     }
 
+    protected function responseIsAnUpgradeResponse($response)
+    {
+        return
+            $response->hasHeader('Connection') &&
+            (in_array('upgrade', array_map('strtolower', $response->getHeader('Connection')))) &&
+            (int) $response->getStatusCode() === 101;
+    }
+
     /** @internal */
     public function handleEnd()
     {

src/Io/DuplexBodyStream.php

@@ -0,0 +1,34 @@
+<?php
+
+namespace React\Http\Io;
+
+use React\Socket\ConnectionInterface;
+use React\Stream\DuplexStreamInterface;
+
+/**
+ * @internal
+ */
+class DuplexBodyStream extends ReadableBodyStream implements DuplexStreamInterface
+{
+    private $connection;
+    public function __construct(ConnectionInterface $connection)
+    {
+        $this->connection = $connection;
+        parent::__construct($connection);
+    }
+
+    public function isWritable()
+    {
+        return $this->connection->isWritable();
+    }
+
+    public function write($data)
+    {
+        return $this->connection->write($data);
+    }
+
+    public function end($data = null)
+    {
+        return $this->connection->end($data);
+    }
+}

src/Io/Transaction.php

@@ -244,15 +244,24 @@ public function onResponse(ResponseInterface $response, RequestInterface $reques
             return $this->onResponseRedirect($response, $request, $deferred, $state);
         }
 
-        // only status codes 200-399 are considered to be valid, reject otherwise
-        if ($this->obeySuccessCode && ($response->getStatusCode() < 200 || $response->getStatusCode() >= 400)) {
-            throw new ResponseException($response);
+        // only status codes 100-399 are considered to be valid, reject otherwise
+        if ($this->obeySuccessCode && $this->failed($response)) {
+            $requestString = \RingCentral\Psr7\str($request);
+            $message = 'HTTP status code ' . $response->getStatusCode() . ' (' . $response->getReasonPhrase() . ')';
+            $message .= ": {$request->getUri()}\n{$requestString}";
+
+            throw new ResponseException($response, $message);
         }
 
         // resolve our initial promise
         return $response;
     }
 
+    private function failed(ResponseInterface $response)
+    {
+        return $response->getStatusCode() >= 500 || ($response->getStatusCode() >= 400 && $response->getStatusCode() < 500);
+    }
+
     /**
      * @param ResponseInterface $response
      * @param RequestInterface $request