Add Request::onInformationalResponseHandler() to process e.g. 103 Ear…

…ly Hints
amphp · Jan 1, 2020 · 098e735 · 098e735
1 parent 4633d5d
commit 098e735
Show file tree

Hide file tree

Showing 3 changed files with 56 additions and 12 deletions.
diff --git a/src/Connection/Http1Connection.php b/src/Connection/Http1Connection.php
@@ -323,6 +323,12 @@ private function readResponse(
                 }
 
                 if ($status < 200) { // 1XX responses (excluding 101, handled above)
+                    $onInformationalResponse = $request->getInformationalResponseHandler();
+
+                    if ($onInformationalResponse !== null) {
+                        yield call($onInformationalResponse, $response);
+                    }
+
                     $chunk = $parser->getBuffer();
                     $parser = new Http1Parser($request, $bodyCallback, $trailersCallback);
                     goto parseChunk;

diff --git a/src/Connection/Internal/Http2ConnectionProcessor.php b/src/Connection/Internal/Http2ConnectionProcessor.php
@@ -2,6 +2,7 @@
 
 namespace Amp\Http\Client\Connection\Internal;
 
+use Amp\ByteStream\InMemoryStream;
 use Amp\ByteStream\IteratorStream;
 use Amp\ByteStream\StreamException;
 use Amp\CancellationToken;
@@ -405,10 +406,6 @@ public function handleHeaders(int $streamId, array $pseudo, array $headers): voi
             return;
         }
 
-        if ($status < 200) {
-            return; // ignore 1xx responses
-        }
-
         asyncCall(function () use ($stream, $streamId) {
             try {
                 foreach ($stream->request->getEventListeners() as $eventListener) {
@@ -419,6 +416,31 @@ public function handleHeaders(int $streamId, array $pseudo, array $headers): voi
             }
         });
 
+        $response = new Response(
+            '2',
+            $status,
+            Status::getReason($status),
+            $headers,
+            new InMemoryStream,
+            $stream->request
+        );
+
+        if ($status < 200) {
+            $onInformationalResponse = $stream->request->getInformationalResponseHandler();
+
+            if ($onInformationalResponse !== null) {
+                asyncCall(function () use ($onInformationalResponse, $response, $streamId) {
+                    try {
+                        yield call($onInformationalResponse, $response);
+                    } catch (\Throwable $e) {
+                        $this->handleStreamException(new Http2StreamException('Informational response handler threw an exception', $streamId, self::CANCEL));
+                    }
+                });
+            }
+
+            return;
+        }
+
         $stream->body = new Emitter;
         $stream->trailers = new Deferred;
 
@@ -428,18 +450,13 @@ public function handleHeaders(int $streamId, array $pseudo, array $headers): voi
             $bodyCancellation->getToken()
         );
 
-        $response = new Response(
-            '2',
-            $status,
-            Status::getReason($status),
-            $headers,
+        $response->setBody(
             new ResponseBodyStream(
                 new IteratorStream($stream->body->iterate()),
                 $bodyCancellation
-            ),
-            $stream->request,
-            $stream->trailers->promise()
+            )
         );
+        $response->setTrailers($stream->trailers->promise());
 
         $stream->responsePending = false;
         $stream->pendingResponse->resolve(call(static function () use ($response, $stream) {

diff --git a/src/Request.php b/src/Request.php
@@ -63,6 +63,9 @@ private static function clone($value)
     /** @var callable|null */
     private $onUpgrade;
 
+    /** @var callable|null */
+    private $onInformationalResponse;
+
     /** @var mixed[] */
     private $attributes = [];
 
@@ -321,6 +324,24 @@ public function getUpgradeHandler(): ?callable
         return $this->onUpgrade;
     }
 
+    /**
+     * Registers a callback invoked when a 1xx response is returned to the request (other than a 101).
+     *
+     * @param callable|null $onInformationalResponse
+     */
+    public function setInformationalResponseHandler(?callable $onInformationalResponse): void
+    {
+        $this->onInformationalResponse = $onInformationalResponse;
+    }
+
+    /**
+     * @return callable|null
+     */
+    public function getInformationalResponseHandler(): ?callable
+    {
+        return $this->onInformationalResponse;
+    }
+
     /**
      * @return int Timeout in milliseconds for the TCP connection.
      */