Make ResponseBody rewindable (#411)

* Make ResponseBody rewindable

* move the cursor at original position while yielding

* Simplify the getChunk method

* Add comment on ResponseBodyStream

* Move resolve on top of stream method

* Fix CS

* Revert getChunk and add documentation

* Document exceptions

Author: jderusse

src/Response.php

@@ -10,6 +10,7 @@
 use AsyncAws\Core\Exception\Http\RedirectionException;
 use AsyncAws\Core\Exception\Http\ServerException;
 use AsyncAws\Core\Exception\RuntimeException;
+use AsyncAws\Core\Stream\ResponseBodyResourceStream;
 use AsyncAws\Core\Stream\ResponseBodyStream;
 use AsyncAws\Core\Stream\ResultStream;
 use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
@@ -129,20 +130,32 @@ public function cancel(): void
         $this->resolveResult = false;
     }
 
+    /**
+     * @throws NetworkException
+     * @throws HttpException
+     */
     public function getHeaders(): array
     {
         $this->resolve();
 
         return $this->httpResponse->getHeaders(false);
     }
 
+    /**
+     * @throws NetworkException
+     * @throws HttpException
+     */
     public function getContent(): string
     {
         $this->resolve();
 
         return $this->httpResponse->getContent(false);
     }
 
+    /**
+     * @throws NetworkException
+     * @throws HttpException
+     */
     public function toArray(): array
     {
         $this->resolve();
@@ -155,8 +168,18 @@ public function getStatusCode(): int
         return $this->httpResponse->getStatusCode();
     }
 
+    /**
+     * @throws NetworkException
+     * @throws HttpException
+     */
     public function toStream(): ResultStream
     {
+        $this->resolve();
+
+        if (\is_callable([$this->httpResponse, 'toStream'])) {
+            return new ResponseBodyResourceStream($this->httpResponse->toStream());
+        }
+
         return new ResponseBodyStream($this->httpClient->stream($this->httpResponse));
     }
 }

src/Stream/ResponseBodyResourceStream.php

@@ -0,0 +1,79 @@
+<?php
+
+declare(strict_types=1);
+
+namespace AsyncAws\Core\Stream;
+
+use AsyncAws\Core\Exception\RuntimeException;
+
+/**
+ * Provides a ResultStream from a resource filled by an HTTP response body.
+ *
+ * @author Jérémy Derussé <[email protected]>
+ */
+class ResponseBodyResourceStream implements ResultStream
+{
+    /**
+     * @var resource
+     */
+    private $resource;
+
+    public function __construct($resource)
+    {
+        $this->resource = $resource;
+    }
+
+    public function __toString()
+    {
+        return $this->getContentAsString();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getChunks(): iterable
+    {
+        $pos = \ftell($this->resource);
+        if (0 !== $pos && !\rewind($this->resource)) {
+            throw new RuntimeException('The stream is not rewindable');
+        }
+
+        try {
+            while (!\feof($this->resource)) {
+                yield \fread($this->resource, 64 * 1024);
+            }
+        } finally {
+            \fseek($this->resource, $pos);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getContentAsString(): string
+    {
+        $pos = \ftell($this->resource);
+
+        try {
+            if (!\rewind($this->resource)) {
+                throw new RuntimeException('Failed to rewind the stream');
+            }
+
+            return \stream_get_contents($this->resource);
+        } finally {
+            \fseek($this->resource, $pos);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getContentAsResource()
+    {
+        if (!\rewind($this->resource)) {
+            throw new RuntimeException('Failed to rewind the stream');
+        }
+
+        return $this->resource;
+    }
+}

src/Stream/ResponseBodyStream.php

@@ -4,12 +4,17 @@
 
 namespace AsyncAws\Core\Stream;
 
+use AsyncAws\Core\Exception\LogicException;
 use Symfony\Contracts\HttpClient\ResponseStreamInterface;
 
 /**
  * Stream a HTTP response body.
+ * This class is a BC layer for Http Response that does not support `toStream()`.
+ * When calling `getChunks` you must read all the chunks before being able to call this method (or another method) again.
+ * When calling `getContentAsResource`, it first, fully read the Response Body in a blocking way.
  *
  * @author Tobias Nyholm <[email protected]>
+ * @author Jérémy Derussé <[email protected]>
  */
 class ResponseBodyStream implements ResultStream
 {
@@ -18,6 +23,13 @@ class ResponseBodyStream implements ResultStream
      */
     private $responseStream;
 
+    /**
+     * @var ResponseBodyResourceStream|null
+     */
+    private $fallback;
+
+    private $partialRead = false;
+
     public function __construct(ResponseStreamInterface $responseStream)
     {
         $this->responseStream = $responseStream;
@@ -33,45 +45,50 @@ public function __toString()
      */
     public function getChunks(): iterable
     {
+        if (null !== $this->fallback) {
+            return $this->fallback->getChunks();
+        }
+        if ($this->partialRead) {
+            throw new LogicException(\sprintf('You can not call "%s". Another process doesn\'t reading "getChunks" till the end.', __METHOD__));
+        }
+
+        $resource = \fopen('php://temp', 'rb+');
         foreach ($this->responseStream as $chunk) {
-            yield $chunk->getContent();
+            $this->partialRead = true;
+            $chunkContent = $chunk->getContent();
+            \fwrite($resource, $chunkContent);
+            yield $chunkContent;
         }
+
+        $this->fallback = new ResponseBodyResourceStream($resource);
+        $this->partialRead = false;
     }
 
     /**
      * {@inheritdoc}
      */
     public function getContentAsString(): string
     {
-        $resource = $this->getContentAsResource();
-
-        try {
-            return \stream_get_contents($resource);
-        } finally {
-            \fclose($resource);
+        if (null === $this->fallback) {
+            // Use getChunks() to read stream content to $this->fallback
+            foreach ($this->getChunks() as $chunk) {
+            }
         }
+
+        return $this->fallback->getContentAsString();
     }
 
     /**
      * {@inheritdoc}
      */
     public function getContentAsResource()
     {
-        $resource = \fopen('php://temp', 'rw+');
-
-        try {
-            foreach ($this->responseStream as $chunk) {
-                fwrite($resource, $chunk->getContent());
+        if (null === $this->fallback) {
+            // Use getChunks() to read stream content to $this->fallback
+            foreach ($this->getChunks() as $chunk) {
             }
-
-            // Rewind
-            \fseek($resource, 0, \SEEK_SET);
-
-            return $resource;
-        } catch (\Throwable $e) {
-            \fclose($resource);
-
-            throw $e;
         }
+
+        return $this->fallback->getContentAsResource();
     }
 }