fix: add streaming flag and error handling

src/Api/ErrorParser/AbstractErrorParser.php

@@ -31,19 +31,6 @@ abstract protected function payload(
         StructureShape $member
     );
 
-    protected function extractPayload(
-        StructureShape $member,
-        ResponseInterface $response
-    ) {
-        if ($member instanceof StructureShape) {
-            // Structure members parse top-level data into a specific key.
-            return $this->payload($response, $member);
-        } else {
-            // Streaming data is just the stream from the response body.
-            return $response->getBody();
-        }
-    }
-
     protected function populateShape(
         array &$data,
         ResponseInterface $response,
@@ -57,16 +44,15 @@ protected function populateShape(
             if (!empty($data['code'])) {
 
                 $errors = $this->api->getOperation($command->getName())->getErrors();
-                foreach ($errors as $key => $error) {
+                foreach ($errors as $error) {
 
                     // If error code matches a known error shape, populate the body
                     if ($this->errorCodeMatches($data, $error)) {
-                        $modeledError = $error;
-                        $data['body'] = $this->extractPayload(
-                            $modeledError,
-                            $response
+                        $data['body'] = $this->payload(
+                            $response,
+                            $error,
                         );
-                        $data['error_shape'] = $modeledError;
+                        $data['error_shape'] = $error;
 
                         foreach ($error->getMembers() as $name => $member) {
                             switch ($member['location']) {

src/Api/ErrorParser/JsonParserTrait.php

@@ -37,10 +37,13 @@ private function genericHandler(ResponseInterface $response): array
             );
         }
 
-        $parsedBody = null;
-        $body = $response->getBody();
-        if (!$body->isSeekable() || $body->getSize()) {
-            $parsedBody = $this->parseJson((string) $body, $response);
+        // We get the full body content
+        $body = $response->getBody()->getContents();
+
+        $parsedBody = [];
+        // Avoid parsing an empty body
+        if (!empty($body)) {
+            $parsedBody = $this->parseJson($body, $response);
         }
 
         // Parse error code from response body
@@ -57,7 +60,7 @@ private function genericHandler(ResponseInterface $response): array
             'code'        => $error_code ?? null,
             'message'     => null,
             'type'        => $error_type,
-            'parsed'      => $parsedBody
+            'parsed'      => $parsedBody,
         ];
     }
 
@@ -132,13 +135,15 @@ protected function payload(
         ResponseInterface $response,
         StructureShape $member
     ) {
-        $body = $response->getBody();
-        if (!$body->isSeekable() || $body->getSize()) {
-            $jsonBody = $this->parseJson($body, $response);
-        } else {
-            $jsonBody = (string) $body;
+        $body = $response->getBody()->getContents();
+
+        // Avoid parsing empty bodies
+        if (empty($body)) {
+            return [];
         }
 
+        $jsonBody = $this->parseJson($body, $response);
+
         return $this->parser->parse($member, $jsonBody);
     }
 }

src/Api/ErrorParser/JsonRpcErrorParser.php

@@ -3,6 +3,7 @@
 
 use Aws\Api\Parser\JsonParser;
 use Aws\Api\Service;
+use Aws\Api\ResponseWrapper;
 use Aws\CommandInterface;
 use Psr\Http\Message\ResponseInterface;
 
@@ -25,23 +26,29 @@ public function __invoke(
         ResponseInterface $response,
         ?CommandInterface $command = null
     ) {
+        $response = new ResponseWrapper($response);
         $data = $this->genericHandler($response);
 
         // Make the casing consistent across services.
-        if ($data['parsed']) {
-            $data['parsed'] = array_change_key_case($data['parsed']);
+        $parsed = $data['parsed'] ?? null;
+        if ($parsed) {
+            $parsed = array_change_key_case($data['parsed']);
         }
 
-        if (isset($data['parsed']['__type'])) {
+        if (isset($parsed['__type'])) {
             if (!isset($data['code'])) {
-                $parts = explode('#', $data['parsed']['__type']);
-                $data['code'] = isset($parts[1]) ? $parts[1] : $parts[0];
+                $parts = explode('#', $parsed['__type']);
+                $data['code'] = $parts[1] ?? $parts[0];
             }
-            $data['message'] = $data['parsed']['message'] ?? null;
+
+            $data['message'] = $parsed['message'] ?? null;
         }
 
         $this->populateShape($data, $response, $command);
 
+        // Now lets make parsed to be all lowercase
+        $data['parsed'] = $parsed;
+
         return $data;
     }
 }

src/Api/ErrorParser/RestJsonErrorParser.php

@@ -3,7 +3,7 @@
 
 use Aws\Api\Parser\JsonParser;
 use Aws\Api\Service;
-use Aws\Api\StructureShape;
+use Aws\Api\ResponseWrapper;
 use Aws\CommandInterface;
 use Psr\Http\Message\ResponseInterface;
 
@@ -26,6 +26,7 @@ public function __invoke(
         ResponseInterface $response,
         ?CommandInterface $command = null
     ) {
+        $response = new ResponseWrapper($response);
         $data = $this->genericHandler($response);
 
         // Merge in error data from the JSON body
@@ -38,9 +39,14 @@ public function __invoke(
             $data['type'] = strtolower($data['type']);
         }
 
+        // Make the casing consistent across services.
+        $parsed = $data['parsed'] ?? null;
+        if ($parsed) {
+            $parsed = array_change_key_case($data['parsed']);
+        }
+
         // Retrieve error message directly
-        $data['message'] = $data['parsed']['message']
-            ?? ($data['parsed']['Message'] ?? null);
+        $data['message'] = $parsed['message'] ?? null;
 
         $this->populateShape($data, $response, $command);
 

src/Api/ErrorParser/XmlErrorParser.php

@@ -5,6 +5,7 @@
 use Aws\Api\Parser\XmlParser;
 use Aws\Api\Service;
 use Aws\Api\StructureShape;
+use Aws\Api\ResponseWrapper;
 use Aws\CommandInterface;
 use Psr\Http\Message\ResponseInterface;
 
@@ -27,18 +28,22 @@ public function __invoke(
         ResponseInterface $response,
         ?CommandInterface $command = null
     ) {
+        $response = new ResponseWrapper($response);
         $code = (string) $response->getStatusCode();
 
         $data = [
             'type' => $code[0] == '4' ? 'client' : 'server',
             'request_id' => null,
             'code' => null,
             'message' => null,
-            'parsed' => null
+            'parsed' => null,
         ];
 
-        $body = $response->getBody();
-        if ($body->getSize() > 0) {
+        // Get the full body content
+        $body = $response->getBody()->getContents();
+
+        // Parse just if is not empty
+        if (!empty($body)) {
             $this->parseBody($this->parseXml($body, $response), $data);
         } else {
             $this->parseHeaders($response, $data);
@@ -100,12 +105,27 @@ protected function payload(
         ResponseInterface $response,
         StructureShape $member
     ) {
-        $xmlBody = $this->parseXml($response->getBody(), $response);
+        $body = $response->getBody()->getContents();
+
+        // Avoid parsing empty bodies
+        if (empty($body)) {
+            return [];
+        }
+
+        $xmlBody = $this->parseXml(
+            $body,
+            $response
+        );
+
         $prefix = $this->registerNamespacePrefix($xmlBody);
         $errorBody = $xmlBody->xpath("//{$prefix}Error");
 
         if (is_array($errorBody) && !empty($errorBody[0])) {
             return $this->parser->parse($member, $errorBody[0]);
         }
+
+        // Fallback since we should either throw an exception or return a value
+        // when the condition above is not met.
+        return [];
     }
 }

src/Api/Operation.php

@@ -89,7 +89,7 @@ public function getOutput()
     /**
      * Get an array of operation error shapes.
      *
-     * @return Shape[]
+     * @return StructureShape[]
      */
     public function getErrors()
     {

src/Api/Parser/AbstractRestParser.php

@@ -2,6 +2,7 @@
 namespace Aws\Api\Parser;
 
 use Aws\Api\DateTimeResult;
+use Aws\Api\ResponseWrapper;
 use Aws\Api\Shape;
 use Aws\Api\StructureShape;
 use Aws\Result;
@@ -55,9 +56,16 @@ public function __invoke(
             }
         }
 
-        $body = $response->getBody();
+        // Wrap response to allow the body
+        // to be read multiple times
+        // even in non-seekable streams.
+        $response = new ResponseWrapper($response);
+
+        // Get the body content to know whether is empty
+        $body = $response->getBody()->getContents();
+        // Make sure empty payloads are not parsed
         if (!$payload
-            && (!$body->isSeekable() || $body->getSize())
+            && !empty($body)
             && count($output->getMembers()) > 0
         ) {
             // if no payload was found, then parse the contents of the body
@@ -75,17 +83,25 @@ private function extractPayload(
     ) {
         $member = $output->getMember($payload);
         $body = $response->getBody();
-
         if (!empty($member['eventstream'])) {
             $result[$payload] = new EventParsingIterator(
                 $body,
                 $member,
                 $this
             );
-        } elseif ($member instanceof StructureShape) {
+
+            return;
+        }
+
+        // We don't create the wrapper above
+        // because on event-stream we want to keep
+        // the original non-seekable stream.
+        $response = new ResponseWrapper($response);
+        if ($member instanceof StructureShape) {
             //Unions must have at least one member set to a non-null value
             // If the body is empty, we can assume it is unset
-            if (!empty($member['union']) && ($body->isSeekable() && !$body->getSize())) {
+            if (!empty($member['union'])
+                && empty($response->getBody()->getContents())) {
                 return;
             }
 

src/Api/Parser/QueryParser.php

@@ -40,8 +40,10 @@ public function __invoke(
         ResponseInterface $response
     ) {
         $output = $this->api->getOperation($command->getName())->getOutput();
-        $body = $response->getBody();
-        $xml = !$body->isSeekable() || $body->getSize()
+        // Read the full payload, even in non-seekable streams
+        $body = $response->getBody()->getContents();
+        // Just parse when the body is not empty
+        $xml = !empty($body)
             ? $this->parseXml($body, $response)
             : null;
 

src/Api/ResponseWrapper.php

@@ -0,0 +1,46 @@
+<?php
+
+namespace Aws\Api;
+
+use GuzzleHttp\Psr7\CachingStream;
+use GuzzleHttp\Psr7\Response;
+use Psr\Http\Message\ResponseInterface;
+use Psr\Http\Message\StreamInterface;
+
+/**
+ * Wrapper response class that makes sure a stream is always
+ * readable since the beginning. When a stream is non-seekable then
+ * we wrapper it into a caching stream that will allow the stream
+ * to rewind.
+ *
+ * @internal
+ */
+final class ResponseWrapper extends Response
+{
+    public function __construct(ResponseInterface $response)
+    {
+        $body = $response->getBody();
+        if (!$body->isSeekable()) {
+            $body = new CachingStream($response->getBody());
+        }
+
+        parent::__construct(
+            $response->getStatusCode(),
+            $response->getHeaders(),
+            $body,
+            $response->getProtocolVersion(),
+            $response->getReasonPhrase()
+        );
+    }
+
+    /**
+     * @return StreamInterface
+     */
+    public function getBody(): StreamInterface
+    {
+        $stream = parent::getBody();
+        $stream->rewind();
+
+        return $stream;
+    }
+}

src/AwsClient.php

@@ -283,6 +283,7 @@ public function __construct(array $args)
             $args['with_resolved']($config);
         }
         $this->addUserAgentMiddleware($config);
+        $this->addEventStreamHttpFlagMiddleware();
     }
 
     public function getHandlerList()
@@ -643,6 +644,33 @@ private function addUserAgentMiddleware($args)
         );
     }
 
+    /**
+     * Enables streaming the response by using the stream flag.
+     *
+     * @return void
+     */
+    private function addEventStreamHttpFlagMiddleware(): void
+    {
+        $this->getHandlerList()
+            -> appendInit(
+                function (callable $handler) {
+                    return function (CommandInterface $command, $request = null) use ($handler) {
+                        $operation = $this->getApi()->getOperation($command->getName());
+                        $output = $operation->getOutput();
+                        foreach ($output->getMembers() as $memberProps) {
+                            if (!empty($memberProps['eventstream'])) {
+                                $command['@http']['stream'] = true;
+                                break;
+                            }
+                        }
+
+                        return $handler($command, $request);
+                    };
+                },
+                'event-streaming-flag-middleware'
+            );
+    }
+
     /**
      * Retrieves client context param definition from service model,
      * creates mapping of client context param names with client-provided