[WIP] Expose DuplexStreamInterface for connections that requires Upgrades

src/Browser.php

@@ -10,6 +10,7 @@
 use React\Http\Io\Sender;
 use React\Http\Io\Transaction;
 use React\Promise\PromiseInterface;
+use React\Socket\ConnectionInterface;
 use React\Socket\ConnectorInterface;
 use React\Stream\ReadableStreamInterface;
 use InvalidArgumentException;
@@ -707,7 +708,7 @@ public function withResponseBuffer($maximumSize)
      * @see self::withFollowRedirects()
      * @see self::withRejectErrorResponse()
      */
-    private function withOptions(array $options)
+    public function withOptions(array $options)
     {
         $browser = clone $this;
         $browser->transaction = $this->transaction->withOptions($options);
@@ -720,7 +721,7 @@ private function withOptions(array $options)
      * @param string                         $url
      * @param array                          $headers
      * @param string|ReadableStreamInterface $body
-     * @return PromiseInterface<ResponseInterface,Exception>
+     * @return PromiseInterface<ResponseInterface,Exception, ConnectionInterface>
      */
     private function requestMayBeStreaming($method, $url, array $headers = array(), $body = '')
     {

src/Client/Request.php

@@ -89,6 +89,14 @@ function (ConnectionInterface $stream) use ($requestData, &$streamRef, &$stateRe
         });
     }
 
+    protected function responseIsAnUpgradeResponse($response)
+    {
+        return
+            $response->hasHeader('Connection') &&
+            (in_array('upgrade', array_map('strtolower', $response->getHeader('Connection')))) &&
+            (int) $response->getStatusCode() === 101;
+    }
+
     public function write($data)
     {
         if (!$this->isWritable()) {
@@ -138,7 +146,16 @@ public function handleData($data)
         // buffer until double CRLF (or double LF for compatibility with legacy servers)
         if (false !== strpos($this->buffer, "\r\n\r\n") || false !== strpos($this->buffer, "\n\n")) {
             try {
-                list($response, $bodyChunk) = $this->parseResponse($this->buffer);
+                $psrResponse = gPsr\parse_response($this->buffer);
+
+                if ($this->responseIsAnUpgradeResponse($psrResponse)) {
+                    $this->stream->removeListener('data', array($this, 'handleData'));
+
+                    $this->emit('upgrade', array($this->stream, $psrResponse, $this));
+                    return;
+                }
+
+                list($response, $bodyChunk) = $this->parseResponse($psrResponse);
             } catch (\InvalidArgumentException $exception) {
                 $this->emit('error', array($exception));
             }
@@ -222,9 +239,8 @@ public function close()
         $this->removeAllListeners();
     }
 
-    protected function parseResponse($data)
+    protected function parseResponse($psrResponse)
     {
-        $psrResponse = gPsr\parse_response($data);
         $headers = array_map(function($val) {
             if (1 === count($val)) {
                 $val = $val[0];
@@ -292,4 +308,9 @@ public function getResponseFactory()
 
         return $factory;
     }
+
+    public function getRequestData()
+    {
+        return $this->requestData;
+    }
 }

src/Client/RequestData.php

@@ -125,4 +125,12 @@ private function getAuthHeaders()
 
         return array();
     }
+
+    /**
+     * @return array
+     */
+    public function getHeaders()
+    {
+        return $this->headers;
+    }
 }

src/Client/UpgradedResponse.php

@@ -0,0 +1,56 @@
+<?php
+
+namespace React\Http\Client;
+
+use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
+use React\Stream\DuplexStreamInterface;
+
+class UpgradedResponse
+{
+    /**
+     * @var DuplexStreamInterface
+     */
+    private $connection;
+
+    /**
+     * @var ResponseInterface
+     */
+    private $response;
+
+    /**
+     * @var RequestInterface
+     */
+    private $request;
+
+    public function __construct(DuplexStreamInterface $connection, ResponseInterface $response, RequestInterface $request)
+    {
+        $this->connection = $connection;
+        $this->response = $response;
+        $this->request = $request;
+    }
+
+    /**
+     * @return DuplexStreamInterface
+     */
+    public function getConnection()
+    {
+        return $this->connection;
+    }
+
+    /**
+     * @return ResponseInterface
+     */
+    public function getResponse()
+    {
+        return $this->response;
+    }
+
+    /**
+     * @return RequestInterface
+     */
+    public function getRequest()
+    {
+        return $this->request;
+    }
+}

src/Io/Sender.php

@@ -3,13 +3,16 @@
 namespace React\Http\Io;
 
 use Psr\Http\Message\RequestInterface;
+use Psr\Http\Message\ResponseInterface;
 use React\EventLoop\LoopInterface;
 use React\Http\Client\Client as HttpClient;
 use React\Http\Client\Response as ResponseStream;
 use React\Http\Message\Response;
+use React\Http\Client\UpgradedResponse;
 use React\Promise\PromiseInterface;
 use React\Promise\Deferred;
 use React\Socket\ConnectorInterface;
+use React\Stream\DuplexStreamInterface;
 use React\Stream\ReadableStreamInterface;
 
 /**
@@ -70,33 +73,14 @@ public function __construct(HttpClient $http)
      *
      * @internal
      * @param RequestInterface $request
-     * @return PromiseInterface Promise<ResponseInterface, Exception>
+     * @param bool $upgrade Configures the sender to listen for upgrade
+     * @return PromiseInterface Promise<ResponseInterface, ConnectionInterface, Exception>
      */
-    public function send(RequestInterface $request)
+    public function send(RequestInterface $request, $upgrade = false)
     {
         $body = $request->getBody();
-        $size = $body->getSize();
-
-        if ($size !== null && $size !== 0) {
-            // automatically assign a "Content-Length" request header if the body size is known and non-empty
-            $request = $request->withHeader('Content-Length', (string)$size);
-        } elseif ($size === 0 && \in_array($request->getMethod(), array('POST', 'PUT', 'PATCH'))) {
-            // only assign a "Content-Length: 0" request header if the body is expected for certain methods
-            $request = $request->withHeader('Content-Length', '0');
-        } elseif ($body instanceof ReadableStreamInterface && $body->isReadable() && !$request->hasHeader('Content-Length')) {
-            // use "Transfer-Encoding: chunked" when this is a streaming body and body size is unknown
-            $request = $request->withHeader('Transfer-Encoding', 'chunked');
-        } else {
-            // do not use chunked encoding if size is known or if this is an empty request body
-            $size = 0;
-        }
-
-        $headers = array();
-        foreach ($request->getHeaders() as $name => $values) {
-            $headers[$name] = implode(', ', $values);
-        }
 
-        $requestStream = $this->http->request($request->getMethod(), (string)$request->getUri(), $headers, $request->getProtocolVersion());
+        list($size, $requestStream) = $this->createRequestStream($request);
 
         $deferred = new Deferred(function ($_, $reject) use ($requestStream) {
             // close request stream if request is cancelled
@@ -127,6 +111,16 @@ public function send(RequestInterface $request)
             ));
         });
 
+        /**
+         * We listen for an upgrade, if the request was upgraded, we will hijack it and resolve immediately
+         * This is useful for websocket connections that requires an HTTP Upgrade.
+         */
+        if ($upgrade) {
+            $requestStream->on('upgrade', function (DuplexStreamInterface $socket, ResponseInterface $response) use ($request, $deferred) {
+                $deferred->resolve(new UpgradedResponse($socket, $response, $request));
+            });
+        }
+
         if ($body instanceof ReadableStreamInterface) {
             if ($body->isReadable()) {
                 // length unknown => apply chunked transfer-encoding
@@ -162,4 +156,31 @@ public function send(RequestInterface $request)
 
         return $deferred->promise();
     }
+
+    protected function createRequestStream(RequestInterface $request)
+    {
+        $body = $request->getBody();
+        $size = $body->getSize();
+
+        if ($size !== null && $size !== 0) {
+            // automatically assign a "Content-Length" request header if the body size is known and non-empty
+            $request = $request->withHeader('Content-Length', (string)$size);
+        } elseif ($size === 0 && \in_array($request->getMethod(), array('POST', 'PUT', 'PATCH'))) {
+            // only assign a "Content-Length: 0" request header if the body is expected for certain methods
+            $request = $request->withHeader('Content-Length', '0');
+        } elseif ($body instanceof ReadableStreamInterface && $body->isReadable() && !$request->hasHeader('Content-Length')) {
+            // use "Transfer-Encoding: chunked" when this is a streaming body and body size is unknown
+            $request = $request->withHeader('Transfer-Encoding', 'chunked');
+        } else {
+            // do not use chunked encoding if size is known or if this is an empty request body
+            $size = 0;
+        }
+
+        $headers = array();
+        foreach ($request->getHeaders() as $name => $values) {
+            $headers[$name] = implode(', ', $values);
+        }
+
+        return array($size, $this->http->request($request->getMethod(), (string)$request->getUri(), $headers, $request->getProtocolVersion()));
+    }
 }

src/Io/Transaction.php

@@ -35,6 +35,9 @@ class Transaction
 
     private $streaming = false;
 
+    // Determines whether to return the connection of an upgrade or not
+    private $upgrade = false;
+
     private $maximumSize = 16777216; // 16 MiB = 2^24 bytes
 
     public function __construct(Sender $sender, LoopInterface $loop)
@@ -81,7 +84,7 @@ public function send(RequestInterface $request)
 
         $loop = $this->loop;
         $this->next($request, $deferred)->then(
-            function (ResponseInterface $response) use ($deferred, $loop, &$timeout) {
+            function ($response) use ($deferred, $loop, &$timeout) {
                 if (isset($deferred->timeout)) {
                     $loop->cancelTimer($deferred->timeout);
                     unset($deferred->timeout);
@@ -144,7 +147,9 @@ private function next(RequestInterface $request, Deferred $deferred)
         $that = $this;
         ++$deferred->numRequests;
 
-        $promise = $this->sender->send($request);
+        $promise = $this->sender->send($request, $this->upgrade);
+
+        if ($this->upgrade) return $promise;
 
         if (!$this->streaming) {
             $promise = $promise->then(function ($response) use ($deferred, $that) {