Fail streams only with StreamException (#354)

amphp · Dec 10, 2023 · fffc8c3 · fffc8c3
1 parent 0347699
commit fffc8c3
Show file tree

Hide file tree

Showing 6 changed files with 41 additions and 13 deletions.
diff --git a/src/Connection/Http1Connection.php b/src/Connection/Http1Connection.php
@@ -432,8 +432,18 @@ private function readResponse(
                     } catch (\Throwable $e) {
                         $this->close();
 
-                        $bodyEmitter->error($e);
+                        $e = $this->wrapException($e);
+
                         $trailersDeferred->error($e);
+
+                        if (!$e instanceof CancelledException) {
+                            $e = new StreamException(
+                                'HTTP response did not complete: ' . $e->getMessage(),
+                                previous: $e,
+                            );
+                        }
+
+                        $bodyEmitter->error($e);
                     } finally {
                         $bodyCancellation->unsubscribe($closeId);
                     }
@@ -690,4 +700,13 @@ public function getConnectDuration(): float
     {
         return $this->connectDuration;
     }
+
+    private function wrapException(\Throwable $exception): \Throwable
+    {
+        if ($exception instanceof HttpException || $exception instanceof CancelledException) {
+            return $exception;
+        }
+
+        return new HttpException($exception->getMessage(), previous: $exception);
+    }
 }
diff --git a/src/Connection/Internal/Http2ConnectionProcessor.php b/src/Connection/Internal/Http2ConnectionProcessor.php
@@ -1356,12 +1356,19 @@ private function releaseStream(int $streamId, ?\Throwable $exception, bool $unpr
             $stream->pendingResponse?->error($exception);
             $stream->pendingResponse = null;
 
-            $stream->body?->error($exception);
-            $stream->body = null;
-
             $stream->trailers?->error($exception);
             $stream->trailers = null;
 
+            if (!$exception instanceof CancelledException) {
+                $exception = new StreamException(
+                    'HTTP response did not complete: ' . $exception->getMessage(),
+                    previous: $exception,
+                );
+            }
+
+            $stream->body?->error($exception);
+            $stream->body = null;
+
             $this->writeFrame(
                 Http2Parser::RST_STREAM,
                 Http2Parser::NO_FLAG,

diff --git a/test/ClientHttpBinIntegrationTest.php b/test/ClientHttpBinIntegrationTest.php
@@ -72,7 +72,7 @@ public function testIncompleteHttpResponseWithContentLength(): void
     {
         $this->givenRawServerResponse("HTTP/1.0 200 OK\r\nContent-Length: 2\r\n\r\n.");
 
-        $this->expectException(SocketException::class);
+        $this->expectException(StreamException::class);
         $this->expectExceptionMessage("Socket disconnected prior to response completion");
 
         $request = $this->createRequest();
@@ -86,7 +86,7 @@ public function testIncompleteHttpResponseWithChunkedEncoding(): void
     {
         $this->givenRawServerResponse("HTTP/1.0 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n0\r"); // missing \n
 
-        $this->expectException(SocketException::class);
+        $this->expectException(StreamException::class);
         $this->expectExceptionMessage("Socket disconnected prior to response completion");
 
         $request = $this->createRequest();

diff --git a/test/Connection/Http1ConnectionTest.php b/test/Connection/Http1ConnectionTest.php
@@ -167,7 +167,7 @@ public function testTransferTimeout(): void
         try {
             $response->getBody()->buffer();
             self::fail("The request should have timed out");
-        } catch (TimeoutException $exception) {
+        } catch (StreamException $exception) {
             self::assertStringContainsString('transfer timeout', $exception->getMessage());
         }
     }
@@ -208,7 +208,7 @@ public function testInactivityTimeout(): void
         try {
             $response->getBody()->buffer();
             self::fail("The request should have timed out");
-        } catch (TimeoutException $exception) {
+        } catch (StreamException $exception) {
             self::assertStringContainsString('Inactivity timeout', $exception->getMessage());
         }
     }

diff --git a/test/Connection/Http2ConnectionTest.php b/test/Connection/Http2ConnectionTest.php
@@ -3,6 +3,7 @@
 
 namespace Amp\Http\Client\Connection;
 
+use Amp\ByteStream\StreamException;
 use Amp\CancelledException;
 use Amp\Future;
 use Amp\Http\Client\HttpException;
@@ -281,7 +282,7 @@ public function testTimeoutWhileStreamingBody(): void
         try {
             $response->getBody()->buffer();
             self::fail("The request body should have been cancelled");
-        } catch (TimeoutException $exception) {
+        } catch (StreamException $exception) {
             delay(0.01); // Allow frame queue to complete writing.
             $buffer = $server->read();
             $expected = \bin2hex(self::packFrame(
@@ -419,7 +420,7 @@ public function testInactivityWhileStreamingBody(): void
         try {
             $response->getBody()->buffer();
             self::fail("The request body should have been cancelled");
-        } catch (TimeoutException $exception) {
+        } catch (StreamException $exception) {
             delay(0.01); // Allow frame queue to complete writing.
             $buffer = $server->read();
             $expected = \bin2hex(self::packFrame(
@@ -485,7 +486,7 @@ public function testServerPushingOddStream(): void
                 [":path", '/static'],
             ]), Http2Parser::PUSH_PROMISE, Http2Parser::END_HEADERS, 1));
 
-        $this->expectException(SocketException::class);
+        $this->expectException(StreamException::class);
         $this->expectExceptionMessage('Invalid server initiated stream');
 
         /** @var Response $response */

diff --git a/test/TimeoutTest.php b/test/TimeoutTest.php
@@ -3,6 +3,7 @@
 
 namespace Amp\Http\Client;
 
+use Amp\ByteStream\StreamException;
 use Amp\Http\Client\Connection\DefaultConnectionFactory;
 use Amp\Http\Client\Connection\UnlimitedConnectionPool;
 use Amp\Http\Client\Interceptor\SetRequestTimeout;
@@ -48,7 +49,7 @@ public function testTimeoutDuringBody(): void
 
             $response = $this->client->request($request);
 
-            $this->expectException(TimeoutException::class);
+            $this->expectException(StreamException::class);
             $this->expectExceptionMessage("Allowed transfer timeout exceeded, took longer than 1 s");
 
             $response->getBody()->buffer();
@@ -172,7 +173,7 @@ public function testTimeoutDuringBodyInterceptor(): void
             $client = new InterceptedHttpClient(new PooledHttpClient, new SetRequestTimeout(10, 10, 1), []);
             $response = $client->request($request, new NullCancellation);
 
-            $this->expectException(TimeoutException::class);
+            $this->expectException(StreamException::class);
             $this->expectExceptionMessage("Allowed transfer timeout exceeded, took longer than 1 s");
 
             $response->getBody()->buffer();