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

@@ -39,8 +39,13 @@ private function genericHandler(ResponseInterface $response): array
 
         $parsedBody = null;
         $body = $response->getBody();
-        if (!$body->isSeekable() || $body->getSize()) {
-            $parsedBody = $this->parseJson((string) $body, $response);
+        if ($body->isSeekable()) {
+            $body->rewind();
+        }
+
+        $rawBody = $body->getContents();
+        if (!empty($rawBody)) {
+            $parsedBody = $this->parseJson($rawBody, $response);
         }
 
         // Parse error code from response body
@@ -133,10 +138,15 @@ protected function payload(
         StructureShape $member
     ) {
         $body = $response->getBody();
-        if (!$body->isSeekable() || $body->getSize()) {
-            $jsonBody = $this->parseJson($body, $response);
+        if ($body->isSeekable()) {
+            $body->rewind();
+        }
+
+        $rawBody = $body->getContents();
+        if (!empty($rawBody)) {
+            $jsonBody = $this->parseJson($rawBody, $response);
         } else {
-            $jsonBody = (string) $body;
+            $jsonBody = $rawBody;
         }
 
         return $this->parser->parse($member, $jsonBody);

src/Api/ErrorParser/JsonRpcErrorParser.php

@@ -4,6 +4,8 @@
 use Aws\Api\Parser\JsonParser;
 use Aws\Api\Service;
 use Aws\CommandInterface;
+use GuzzleHttp\Psr7\CachingStream;
+use GuzzleHttp\Psr7\Response;
 use Psr\Http\Message\ResponseInterface;
 
 /**
@@ -25,6 +27,15 @@ public function __invoke(
         ResponseInterface $response,
         ?CommandInterface $command = null
     ) {
+        if (!$response->getBody()->isSeekable()) {
+            $response = new Response(
+                $response->getStatusCode(),
+                $response->getHeaders(),
+                new CachingStream($response->getBody()),
+                $response->getProtocolVersion(),
+                $response->getReasonPhrase()
+            );
+        }
         $data = $this->genericHandler($response);
 
         // Make the casing consistent across services.

src/Api/ErrorParser/XmlErrorParser.php

@@ -100,12 +100,24 @@ protected function payload(
         ResponseInterface $response,
         StructureShape $member
     ) {
-        $xmlBody = $this->parseXml($response->getBody(), $response);
+        $body = $response->getBody();
+        if ($body->isSeekable()) {
+            $body->rewind();
+        }
+
+        $rawBody = $body->getContents();
+        if (empty($rawBody)) {
+            return $rawBody;
+        }
+
+        $xmlBody = $this->parseXml($rawBody, $response);
         $prefix = $this->registerNamespacePrefix($xmlBody);
         $errorBody = $xmlBody->xpath("//{$prefix}Error");
 
         if (is_array($errorBody) && !empty($errorBody[0])) {
             return $this->parser->parse($member, $errorBody[0]);
         }
+
+        return $rawBody;
     }
 }

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,10 +2,13 @@
 namespace Aws\Api\Parser;
 
 use Aws\Api\DateTimeResult;
+use Aws\Api\ResponseWrapper;
 use Aws\Api\Shape;
 use Aws\Api\StructureShape;
 use Aws\Result;
 use Aws\CommandInterface;
+use GuzzleHttp\Psr7\CachingStream;
+use GuzzleHttp\Psr7\Response;
 use Psr\Http\Message\ResponseInterface;
 
 /**
@@ -39,6 +42,30 @@ public function __invoke(
 
         if ($payload = $output['payload']) {
             $this->extractPayload($payload, $output, $response, $result);
+        } else {
+            $body = $response->getBody();
+            if (!$body->isSeekable()) {
+                $response = new Response(
+                    $response->getStatusCode(),
+                    $response->getHeaders(),
+                    new CachingStream($body),
+                    $response->getProtocolVersion(),
+                    $response->getReasonPhrase()
+                );
+            }
+
+            $body = $response->getBody();
+            if ($body->isSeekable()) {
+                $body->rewind();
+            }
+
+            $rawBody = $body->getContents();
+            if (!empty($rawBody)
+                && count($output->getMembers()) > 0
+            ) {
+                // if no payload was found, then parse the contents of the body
+                $this->payload($response, $output, $result);
+            }
         }
 
         foreach ($output->getMembers() as $name => $member) {
@@ -55,15 +82,6 @@ public function __invoke(
             }
         }
 
-        $body = $response->getBody();
-        if (!$payload
-            && (!$body->isSeekable() || $body->getSize())
-            && count($output->getMembers()) > 0
-        ) {
-            // if no payload was found, then parse the contents of the body
-            $this->payload($response, $output, $result);
-        }
-
         return new Result($result);
     }
 
@@ -75,17 +93,37 @@ 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;
+        }
+
+        if (!$body->isSeekable()) {
+            $response = new Response(
+                $response->getStatusCode(),
+                $response->getHeaders(),
+                new CachingStream($response->getBody()),
+                $response->getProtocolVersion(),
+                $response->getReasonPhrase()
+            );
+        }
+
+        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())) {
+            $body = $response->getBody();
+            if ($body->isSeekable()) {
+                $body->rewind();
+            }
+
+            $rawBody = $body->getContents();
+            if (!empty($member['union'])
+                && empty($rawBody)) {
                 return;
             }
 

src/Api/Parser/JsonRpcParser.php

@@ -63,11 +63,16 @@ private function parseResponse(ResponseInterface $response, Operation $operation
             }
         }
 
+        $body = $response->getBody();
+        if ($body->isSeekable()) {
+            $body->rewind();
+        }
+
         $result = $this->parseMemberFromStream(
-                $response->getBody(),
-                $operation->getOutput(),
-                $response
-            );
+            $body,
+            $operation->getOutput(),
+            $response
+        );
 
         return new Result(is_null($result) ? [] : $result);
     }

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/Parser/RestJsonParser.php

@@ -28,15 +28,19 @@ protected function payload(
         StructureShape $member,
         array &$result
     ) {
-        $responseBody = (string) $response->getBody();
+        $body = $response->getBody();
+        if ($body->isSeekable()) {
+            $body->rewind();
+        }
+
+        $rawBody = $body->getContents();
 
         // Parse JSON if we have content
-        $parsedJson = null;
-        if (!empty($responseBody)) {
-            $parsedJson = $this->parseJson($responseBody, $response);
+        if (!empty($rawBody)) {
+            $parsedJson = $this->parseJson($rawBody, $response);
         } else {
             // An empty response body should be deserialized as null
-            $result = $parsedJson;
+            $result = null;
             return;
         }
 

src/Api/Parser/RestXmlParser.php

@@ -28,7 +28,12 @@ protected function payload(
         StructureShape $member,
         array &$result
     ) {
-        $result += $this->parseMemberFromStream($response->getBody(), $member, $response);
+        $body = $response->getBody();
+        if ($body->isSeekable()) {
+            $body->rewind();
+        }
+
+        $result += $this->parseMemberFromStream($body, $member, $response);
     }
 
     public function parseMemberFromStream(

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

src/CloudWatchLogs/CloudWatchLogsClient.php

@@ -192,36 +192,10 @@
  * @method \GuzzleHttp\Promise\Promise updateLogAnomalyDetectorAsync(array $args = [])
  */
 class CloudWatchLogsClient extends AwsClient {
-    static $streamingCommands = [
-        'StartLiveTail' => true
-    ];
 
     public function __construct(array $args)
     {
         parent::__construct($args);
-        $this->addStreamingFlagMiddleware();
-    }
-
-    private function addStreamingFlagMiddleware()
-    {
-        $this->getHandlerList()
-            -> appendInit(
-                $this->getStreamingFlagMiddleware(),
-                'streaming-flag-middleware'
-            );
-    }
-
-    private function getStreamingFlagMiddleware(): callable
-    {
-        return function (callable $handler) {
-            return function (CommandInterface $command, $request = null) use ($handler) {
-                if (!empty(self::$streamingCommands[$command->getName()])) {
-                    $command['@http']['stream'] = true;
-                }
-
-                return $handler($command, $request);
-            };
-        };
     }
 
     /**