Move StreamableBody to AsyncAws\Core\Stream (#348)

* Move StreamableBody to AsyncAws\Core\Stream

* Rebase fix

* Bugfix

* minor

Author: Nyholm

src/Request.php

@@ -3,7 +3,7 @@
 namespace AsyncAws\Core;
 
 use AsyncAws\Core\Exception\LogicException;
-use AsyncAws\Core\Stream\Stream;
+use AsyncAws\Core\Stream\RequestStream;
 
 /**
  * Representation of an HTTP Request.
@@ -31,7 +31,7 @@ class Request
     /**
      * @param string[]|string[][] $headers
      */
-    public function __construct(string $method, string $uri, array $query, array $headers, Stream $body)
+    public function __construct(string $method, string $uri, array $query, array $headers, RequestStream $body)
     {
         $this->method = $method;
         $this->uri = $uri;
@@ -87,12 +87,12 @@ public function removeHeader(string $name): void
         unset($this->headers[strtolower($name)]);
     }
 
-    public function getBody(): Stream
+    public function getBody(): RequestStream
     {
         return $this->body;
     }
 
-    public function setBody(Stream $body)
+    public function setBody(RequestStream $body)
     {
         $this->body = $body;
     }

src/Response.php

@@ -10,6 +10,8 @@
 use AsyncAws\Core\Exception\Http\RedirectionException;
 use AsyncAws\Core\Exception\Http\ServerException;
 use AsyncAws\Core\Exception\RuntimeException;
+use AsyncAws\Core\Stream\ResponseBodyStream;
+use AsyncAws\Core\Stream\ResultStream;
 use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
 use Symfony\Contracts\HttpClient\HttpClientInterface;
 use Symfony\Contracts\HttpClient\ResponseInterface;
@@ -151,8 +153,8 @@ public function getStatusCode(): int
         return $this->response->getStatusCode();
     }
 
-    public function toStream(): StreamableBody
+    public function toStream(): ResultStream
     {
-        return new StreamableBody($this->httpClient->stream($this->response));
+        return new ResponseBodyStream($this->httpClient->stream($this->response));
     }
 }

src/Signer/SignerV4.php

@@ -8,7 +8,7 @@
 use AsyncAws\Core\RequestContext;
 use AsyncAws\Core\Stream\FixedSizeStream;
 use AsyncAws\Core\Stream\IterableStream;
-use AsyncAws\Core\Stream\Stream;
+use AsyncAws\Core\Stream\RequestStream;
 use AsyncAws\Core\Stream\StringStream;
 
 /**
@@ -263,7 +263,7 @@ private function convertBodyToStream(Request $request, \DateTimeInterface $now,
         $metaLength = \strlen(";chunk-signature=\r\n\r\n") + 64;
         $request->setHeader('content-length', $contentLength + $fullChunkCount * ($metaLength + \strlen((string) dechex(self::CHUNK_SIZE))) + ($chunkCount - $fullChunkCount) * ($metaLength + \strlen((string) dechex($contentLength % self::CHUNK_SIZE))) + $metaLength + 1);
 
-        $body = IterableStream::create((function (Stream $body) use ($now, $credentialString, $signingKey, &$signature): iterable {
+        $body = IterableStream::create((function (RequestStream $body) use ($now, $credentialString, $signingKey, &$signature): iterable {
             foreach (FixedSizeStream::create($body, self::CHUNK_SIZE) as $chunk) {
                 $stringToSign = $this->buildChunkStringToSign($now, $credentialString, $signature, $chunk);
                 $signature = $this->buildSignature($stringToSign, $signingKey);

src/Stream/CallableStream.php

@@ -12,7 +12,7 @@
  *
  * @author Jérémy Derussé <[email protected]>
  */
-final class CallableStream implements Stream
+final class CallableStream implements RequestStream
 {
     private $content;
 

src/Stream/FixedSizeStream.php

@@ -11,13 +11,13 @@
  *
  * @author Jérémy Derussé <[email protected]>
  */
-final class FixedSizeStream implements Stream
+final class FixedSizeStream implements RequestStream
 {
     private $content;
 
     private $chunkSize;
 
-    private function __construct(Stream $content, int $chunkSize = 64 * 1024)
+    private function __construct(RequestStream $content, int $chunkSize = 64 * 1024)
     {
         $this->content = $content;
         $this->chunkSize = $chunkSize;
@@ -28,11 +28,11 @@ public static function create($content, int $chunkSize = 64 * 1024): FixedSizeSt
         if ($content instanceof self) {
             return $content;
         }
-        if ($content instanceof Stream) {
+        if ($content instanceof RequestStream) {
             return new self($content, $chunkSize);
         }
 
-        throw new InvalidArgument(sprintf('Expect content to be an instance of "%s". "%s" given.', Stream::class, \is_object($content) ? \get_class($content) : \gettype($content)));
+        throw new InvalidArgument(sprintf('Expect content to be an instance of "%s". "%s" given.', RequestStream::class, \is_object($content) ? \get_class($content) : \gettype($content)));
     }
 
     public function length(): ?int

src/Stream/IterableStream.php

@@ -11,7 +11,7 @@
  *
  * @author Jérémy Derussé <[email protected]>
  */
-final class IterableStream implements Stream
+final class IterableStream implements RequestStream
 {
     private $content;
 

src/Stream/RequestStream.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace AsyncAws\Core\Stream;
+
+/**
+ * Provides method to convert a input into string or chunks.
+ *
+ * @internal
+ *
+ * @author Jérémy Derussé <[email protected]>
+ */
+interface RequestStream extends \IteratorAggregate
+{
+    /**
+     * Length in bytes.
+     */
+    public function length(): ?int;
+
+    public function stringify(): string;
+
+    public function hash(string $algo = 'sha256', bool $raw = false): string;
+}

src/Stream/ResourceStream.php

@@ -11,7 +11,7 @@
  *
  * @author Jérémy Derussé <[email protected]>
  */
-final class ResourceStream implements Stream
+final class ResourceStream implements RequestStream
 {
     /**
      * @var resource

src/Stream/ResponseBodyStream.php

@@ -0,0 +1,77 @@
+<?php
+
+declare(strict_types=1);
+
+namespace AsyncAws\Core\Stream;
+
+use Symfony\Contracts\HttpClient\ResponseStreamInterface;
+
+/**
+ * Stream a HTTP response body.
+ *
+ * @author Tobias Nyholm <[email protected]>
+ */
+class ResponseBodyStream implements ResultStream
+{
+    /**
+     * @var ResponseStreamInterface
+     */
+    private $responseStream;
+
+    public function __construct(ResponseStreamInterface $responseStream)
+    {
+        $this->responseStream = $responseStream;
+    }
+
+    public function __toString()
+    {
+        return $this->getContentAsString();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getChunks(): iterable
+    {
+        foreach ($this->responseStream as $chunk) {
+            yield $chunk->getContent();
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getContentAsString(): string
+    {
+        $resource = $this->getContentAsResource();
+
+        try {
+            return \stream_get_contents($resource);
+        } finally {
+            \fclose($resource);
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getContentAsResource()
+    {
+        $resource = \fopen('php://temp', 'rw+');
+
+        try {
+            foreach ($this->responseStream as $chunk) {
+                fwrite($resource, $chunk->getContent());
+            }
+
+            // Rewind
+            \fseek($resource, 0, \SEEK_SET);
+
+            return $resource;
+        } catch (\Throwable $e) {
+            \fclose($resource);
+
+            throw $e;
+        }
+    }
+}

src/Stream/ResultStream.php

@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace AsyncAws\Core\Stream;
+
+use AsyncAws\Core\StreamableBodyInterface;
+
+/**
+ * Stream for a Result.
+ */
+interface ResultStream extends StreamableBodyInterface
+{
+}