Support WebSocket connections

app/Client/Http/HttpClient.php

@@ -54,8 +54,8 @@ public function performRequest(string $requestData, WebSocket $proxyConnection =
 
         $request = $this->passRequestThroughModifiers(parse_request($requestData), $proxyConnection);
 
-        transform($request, function ($request) use ($proxyConnection) {
-            $this->sendRequestToApplication($request, $proxyConnection);
+        return transform($request, function ($request) use ($proxyConnection) {
+            return $this->sendRequestToApplication($request, $proxyConnection);
         });
     }
 
@@ -91,7 +91,7 @@ protected function sendRequestToApplication(RequestInterface $request, $proxyCon
             $uri = $uri->withScheme('https');
         }
 
-        (new Browser($this->loop, $this->createConnector()))
+        return (new Browser($this->loop, $this->createConnector()))
             ->withFollowRedirects(false)
             ->withRejectErrorResponse(false)
             ->requestStreaming(
@@ -109,22 +109,25 @@ protected function sendRequestToApplication(RequestInterface $request, $proxyCon
 
                 $this->sendChunkToServer($responseBuffer, $proxyConnection);
 
-                /* @var $body \React\Stream\ReadableStreamInterface */
+                /* @var $body \React\Stream\DuplexStreamInterface */
                 $body = $response->getBody();
-
                 $this->logResponse(Message::toString($response));
 
-                $body->on('data', function ($chunk) use ($proxyConnection, &$responseBuffer) {
-                    $responseBuffer .= $chunk;
+                if (! $body->isWritable()) {
+                    $body->on('data', function ($chunk) use ($proxyConnection, &$responseBuffer) {
+                        $responseBuffer .= $chunk;
 
-                    $this->sendChunkToServer($chunk, $proxyConnection);
-                });
+                        $this->sendChunkToServer($chunk, $proxyConnection);
+                    });
+                }
 
                 $body->on('close', function () use ($proxyConnection, &$responseBuffer) {
                     $this->logResponse($responseBuffer);
 
                     optional($proxyConnection)->close();
                 });
+
+                return $response;
             });
     }
 

app/Client/ProxyManager.php

@@ -3,6 +3,7 @@
 namespace App\Client;
 
 use App\Client\Http\HttpClient;
+use Psr\Http\Message\ResponseInterface;
 use Ratchet\Client\WebSocket;
 use Ratchet\RFC6455\Messaging\Frame;
 use React\EventLoop\LoopInterface;
@@ -32,8 +33,28 @@ public function createProxy(string $clientId, $connectionData)
             'X-Expose-Control' => 'enabled',
         ], $this->loop)
             ->then(function (WebSocket $proxyConnection) use ($clientId, $connectionData) {
-                $proxyConnection->on('message', function ($message) use ($proxyConnection, $connectionData) {
-                    $this->performRequest($proxyConnection, (string) $message, $connectionData);
+                $localRequestConnection = null;
+
+                $proxyConnection->on('message', function ($message) use (&$localRequestConnection, $proxyConnection, $connectionData) {
+                    if ($localRequestConnection) {
+                        $localRequestConnection->write($message);
+                        return;
+                    }
+
+                    $this->performRequest($proxyConnection, (string) $message, $connectionData)->then(function (ResponseInterface $response) use ($proxyConnection, &$localRequestConnection) {
+                        /** @var $body \React\Stream\DuplexStreamInterface */
+                        $body = $response->getBody();
+                        if ($body) {
+                            $localRequestConnection = $body;
+                        }
+
+                        if ($body->isWritable()) {
+                            $body->on('data', function ($chunk) use ($proxyConnection) {
+                                $binaryMsg = new Frame($chunk, true, Frame::OP_BINARY);
+                                $proxyConnection->send($binaryMsg);
+                            });
+                        }
+                    });
                 });
 
                 $proxyConnection->send(json_encode([
@@ -79,6 +100,6 @@ public function createTcpProxy(string $clientId, $connectionData)
 
     protected function performRequest(WebSocket $proxyConnection, string $requestData, $connectionData)
     {
-        app(HttpClient::class)->performRequest((string) $requestData, $proxyConnection, $connectionData);
+        return app(HttpClient::class)->performRequest((string) $requestData, $proxyConnection, $connectionData);
     }
 }

app/Server/Connections/HttpConnection.php

@@ -30,4 +30,9 @@ public function close()
 
         $this->socket->close();
     }
+
+    public function getConnection(): ConnectionInterface
+    {
+        return $this->socket;
+    }
 }

app/Server/Http/Controllers/TunnelMessageController.php

@@ -91,13 +91,17 @@ protected function sendRequestToClient(Request $request, ControlConnection $cont
 
         $httpConnection = $this->connectionManager->storeHttpConnection($httpConnection, $requestId);
 
-        transform($this->passRequestThroughModifiers($request, $httpConnection), function (Request $request) use ($controlConnection, $requestId) {
-            $controlConnection->once('proxy_ready_'.$requestId, function (ConnectionInterface $proxy) use ($request) {
+        transform($this->passRequestThroughModifiers($request, $httpConnection), function (Request $request) use ($httpConnection, $controlConnection, $requestId) {
+            $controlConnection->once('proxy_ready_'.$requestId, function (ConnectionInterface $proxy) use ($httpConnection, $request) {
                 // Convert the Laravel request into a PSR7 request
                 $psr17Factory = new Psr17Factory();
                 $psrHttpFactory = new PsrHttpFactory($psr17Factory, $psr17Factory, $psr17Factory, $psr17Factory);
                 $request = $psrHttpFactory->createRequest($request);
 
+                $httpConnection->getConnection()->on('data', function($d) use ($proxy) {
+                    $proxy->send(new Frame($d, true, Frame::OP_BINARY));
+                });
+
                 $binaryMsg = new Frame(str($request), true, Frame::OP_BINARY);
                 $proxy->send($binaryMsg);
             });

composer.json

@@ -38,7 +38,7 @@
         "nyholm/psr7": "^1.3",
         "phpunit/phpunit": "^9.4.3",
         "ratchet/pawl": "^0.3.5",
-        "react/http": "^1.1.0",
+        "react/http": "dev-writable-updated",
         "react/socket": "^1.6",
         "react/stream": "^1.1.1",
         "riverline/multipart-parser": "^2.0",
@@ -77,6 +77,10 @@
         {
             "type": "git",
             "url": "https://github.com/filkaris/cuzzle"
+        },
+        {
+            "type": "git",
+            "url": "https://github.com/phpsandbox/http"
         }
     ],
     "minimum-stability": "dev",