Avoid hanging requests in case a request body is closed prematurely (#380)

Author: danog

src/Driver/Http1Driver.php

@@ -882,6 +882,8 @@ private function send(?Future $lastWrite, Response $response, ?Request $request
 
         $chunk = "HTTP/$protocol $status $reason\r\n" . Rfc7230::formatHeaders($headers) . "\r\n";
 
+        $need = $headers["content-length"] ?? null;
+        $wrote = 0;
         try {
             $this->writableStream->write($chunk);
 
@@ -891,6 +893,7 @@ private function send(?Future $lastWrite, Response $response, ?Request $request
                 if ($shouldClose) {
                     $this->writableStream->end();
                 }
+                $need = null;
 
                 return;
             }
@@ -905,6 +908,7 @@ private function send(?Future $lastWrite, Response $response, ?Request $request
                 if ($chunk === "") {
                     continue;
                 }
+                $wrote += strlen($chunk);
 
                 if ($chunked) {
                     $chunk = \sprintf("%x\r\n%s\r\n", \strlen($chunk), $chunk);
@@ -935,7 +939,7 @@ private function send(?Future $lastWrite, Response $response, ?Request $request
             return; // Client will be closed in finally.
         } finally {
             /** @psalm-suppress TypeDoesNotContainType */
-            if ($chunk !== null) {
+            if ($chunk !== null || ($need !== null && $wrote !== (int) $need)) {
                 $this->client->close();
             }
         }

src/Driver/Http2Driver.php

@@ -286,6 +286,9 @@ private function send(int $id, Response $response, Request $request, Cancellatio
     {
         $chunk = ""; // Required for the finally, not directly overwritten, even if your IDE says otherwise.
 
+        $need = $response->getHeader("content-length");
+        $wrote = 0;
+
         try {
             $status = $response->getStatus();
 
@@ -333,6 +336,7 @@ private function send(int $id, Response $response, Request $request, Cancellatio
             if ($request->getMethod() === "HEAD") {
                 $this->streams[$id]->state |= Http2Stream::LOCAL_CLOSED;
                 $this->writeData("", $id);
+                $need = null;
                 return;
             }
 
@@ -344,6 +348,7 @@ private function send(int $id, Response $response, Request $request, Cancellatio
                 if (!isset($this->streams[$id])) {
                     return;
                 }
+                $wrote += strlen($chunk);
 
                 $this->writeData($chunk, $id);
 
@@ -390,7 +395,7 @@ private function send(int $id, Response $response, Request $request, Cancellatio
                 return;
             }
 
-            if ($chunk !== null) {
+            if ($chunk !== null || ($need !== null && $wrote !== (int) $need)) {
                 $error ??= Http2Parser::INTERNAL_ERROR;
                 $this->writeFrame(\pack("N", $error), Http2Parser::RST_STREAM, Http2Parser::NO_FLAG, $id);
                 $this->releaseStream($id, $exception ?? new ClientException($this->client, "Stream error", $error));