wip

beyondcode · Jul 21, 2024 · 28308ab · 28308ab
1 parent 5157e76
commit 28308ab
Show file tree

Hide file tree

Showing 5 changed files with 50 additions and 11 deletions.
diff --git a/app/Client/Http/HttpClient.php b/app/Client/Http/HttpClient.php
@@ -55,7 +55,7 @@ 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 $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;
             });
     }
 

diff --git a/app/Client/ProxyManager.php b/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,30 @@ 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) {
+                $localRequestConnection = null;
+
+                $proxyConnection->on('message', function ($message) use (&$localRequestConnection, $proxyConnection, $connectionData) {
                     $this->performRequest($proxyConnection, (string) $message, $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 +102,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);
     }
 }
diff --git a/app/Server/Connections/HttpConnection.php b/app/Server/Connections/HttpConnection.php
@@ -30,4 +30,9 @@ public function close()
 
         $this->socket->close();
     }
+
+    public function getConnection(): ConnectionInterface
+    {
+        return $this->socket;
+    }
 }
diff --git a/app/Server/Http/Controllers/TunnelMessageController.php b/app/Server/Http/Controllers/TunnelMessageController.php
@@ -98,6 +98,10 @@ protected function sendRequestToClient(Request $request, ControlConnection $cont
                 $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);
             });

diff --git a/composer.json b/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",