feat: add non-buffered response streaming

docs/clients/s3.md

@@ -73,8 +73,10 @@ request which could have a performance impact.
 ### Download files
 
 When you download a file from S3, AsyncAws gives you a `ResultStream` which
-can be used as a string, as a resource, or iterated over. This allows you to handle
-larger files without having them in memory.
+can be used as a string, as a resource, or iterated over. By default, response
+bodies are buffered into temporary storage so the stream can be replayed and
+safely reused by another request. This avoids loading large objects into PHP
+memory, but large objects can still use local temporary storage.
 
 ```
 // download a file and use it directly as string
@@ -104,6 +106,31 @@ foreach ($result->getBody()->getChunks() as $chunk) {
 }
 ```
 
+#### Non-buffered downloads
+
+For one-pass processing of large objects, disable response buffering:
+
+```php
+$result = $s3->getObject([
+    'Bucket' => 'my-company-website',
+    'Key' => 'orders.csv',
+    '@responseBuffer' => false,
+]);
+
+$input = $result->getBody()->getContentAsResource();
+$output = fopen('/path/to/orders.csv', 'wb');
+
+stream_copy_to_stream($input, $output);
+fclose($input);
+fclose($output);
+```
+
+Non-buffered response streams are one-shot and non-rewindable. In this mode,
+`getChunks()` yields chunks without also copying them into `php://temp`. Do not
+pass a non-buffered response stream as the body of another AsyncAWS or Symfony
+HttpClient request. If you need a replayable stream or want to upload a
+downloaded object to another request, keep the default buffered behavior.
+
 ### Add tags to a bucket
 
 You can add tags to your buckets to help you find related resources in the AWS cost explorer; eg all AWS resources tagged

src/CodeGenerator/src/Definition/Operation.php

@@ -112,6 +112,21 @@ public function getOutput(): ?StructureShape
         return null;
     }
 
+    public function hasStreamingOutputPayload(): bool
+    {
+        $output = $this->getOutput();
+        if (null === $output) {
+            return false;
+        }
+
+        $payload = $output->getPayload();
+        if (null === $payload) {
+            return false;
+        }
+
+        return $output->getMember($payload)->isStreaming();
+    }
+
     /**
      * @return ExceptionShape[]
      */

src/CodeGenerator/src/Generator/InputGenerator.php

@@ -116,6 +116,9 @@ public function generate(Operation $operation): ClassName
             if ('region' === $member->getName()) {
                 throw new \RuntimeException('Member conflict with "@region" parameter.');
             }
+            if ('responseBuffer' === $member->getName()) {
+                throw new \RuntimeException('Member conflict with "@responseBuffer" parameter.');
+            }
             $memberShape = $member->getShape();
             [$returnType, $parameterType, $memberClassNames] = $this->typeGenerator->getPhpType($memberShape);
             foreach ($memberClassNames as $memberClassName) {
@@ -280,15 +283,20 @@ public function generate(Operation $operation): ClassName
             ->setReturnType('self')
             ->setBody('return $input instanceof self ? $input : new self($input);');
         $createMethod->addParameter('input');
-        [$doc, $memberClassNames] = $this->typeGenerator->generateDocblock($shape, $className, true, true, false, ['  \'@region\'?: string|null,']);
+        $pseudoOptions = ['  \'@region\'?: string|null,', '  \'@responseBuffer\'?: bool,'];
+        if (0 !== strpos($classBuilder->getClassName()->getFqdn(), 'AsyncAws\\Core\\')) {
+            $this->requirementsRegistry->addRequirement('async-aws/core', '^1.30');
+        }
+
+        [$doc, $memberClassNames] = $this->typeGenerator->generateDocblock($shape, $className, true, true, false, $pseudoOptions);
         $createMethod->addComment($doc);
         foreach ($memberClassNames as $memberClassName) {
             $classBuilder->addUse($memberClassName->getFqdn());
         }
 
         $constructorBody .= 'parent::__construct($input);';
         $constructor = $classBuilder->addMethod('__construct');
-        [$doc, $memberClassNames] = $this->typeGenerator->generateDocblock($shape, $className, false, true, false, ['  \'@region\'?: string|null,']);
+        [$doc, $memberClassNames] = $this->typeGenerator->generateDocblock($shape, $className, false, true, false, $pseudoOptions);
         $constructor->addComment($doc);
         foreach ($memberClassNames as $memberClassName) {
             $classBuilder->addUse($memberClassName->getFqdn());

src/CodeGenerator/src/Generator/OperationGenerator.php

@@ -118,7 +118,12 @@ public function generate(Operation $operation): void
         $method->addComment('@see ' . $operation->getApiReferenceDocumentationUrl());
         $prefix = $operation->getService()->getEndpointPrefix();
         $method->addComment('@see https://docs.aws.amazon.com/aws-sdk-php/v3/api/api-' . $prefix . '-' . $operation->getService()->getApiVersion() . '.html#' . strtolower($operation->getName()));
-        [$doc, $memberClassNames] = $this->typeGenerator->generateDocblock($inputShape, $inputClass, true, false, false, ['  \'@region\'?: string|null,']);
+        $pseudoOptions = ['  \'@region\'?: string|null,'];
+        if ($operation->hasStreamingOutputPayload()) {
+            $pseudoOptions[] = '  \'@responseBuffer\'?: bool,';
+        }
+
+        [$doc, $memberClassNames] = $this->typeGenerator->generateDocblock($inputShape, $inputClass, true, false, false, $pseudoOptions);
         $method->addComment($doc);
         foreach ($memberClassNames as $memberClassName) {
             $classBuilder->addUse($memberClassName->getFqdn());
@@ -206,6 +211,14 @@ private function setMethodBody(Method $method, Operation $operation, ClassName $
             $extra .= ", 'exceptionMapping' => [\n" . implode("\n", $mapping) . "\n]";
         }
 
+        if ($operation->hasStreamingOutputPayload()) {
+            if (0 !== strpos($classBuilder->getClassName()->getFqdn(), 'AsyncAws\\Core\\')) {
+                $this->requirementsRegistry->addRequirement('async-aws/core', '^1.30');
+            }
+
+            $extra .= ", 'responseBuffer' => \$input->shouldBufferResponse()";
+        }
+
         if ($operation->requiresEndpointDiscovery()) {
             if (0 !== strpos($classBuilder->getClassName()->getFqdn(), 'AsyncAws\Core\\')) {
                 $this->requirementsRegistry->addRequirement('async-aws/core', '^1.16');

src/Core/src/AbstractApi.php

@@ -161,12 +161,18 @@ final protected function getResponse(Request $request, ?RequestContext $context
             $requestBody = $requestBody->stringify();
         }
 
+        $responseBuffer = $context ? $context->shouldBufferResponse() : true;
+        $options = [
+            'headers' => $request->getHeaders(),
+        ] + (0 === $length ? [] : ['body' => $requestBody]);
+        if (!$responseBuffer) {
+            $options['buffer'] = false;
+        }
+
         $response = $this->httpClient->request(
             $request->getMethod(),
             $request->getEndpoint(),
-            [
-                'headers' => $request->getHeaders(),
-            ] + (0 === $length ? [] : ['body' => $requestBody])
+            $options
         );
 
         if ($debug = filter_var($this->configuration->get('debug'), \FILTER_VALIDATE_BOOLEAN)) {
@@ -178,7 +184,7 @@ final protected function getResponse(Request $request, ?RequestContext $context
             ]);
         }
 
-        return new Response($response, $this->httpClient, $this->logger, $this->awsErrorFactory, $this->endpointCache, $request, $debug, $context ? $context->getExceptionMapping() : []);
+        return new Response($response, $this->httpClient, $this->logger, $this->awsErrorFactory, $this->endpointCache, $request, $debug, $context ? $context->getExceptionMapping() : [], $responseBuffer);
     }
 
     /**

src/Core/src/Input.php

@@ -15,11 +15,17 @@ abstract class Input
     public $region;
 
     /**
-     * @param array{'@region'?: ?string,...} $input
+     * @var bool
+     */
+    private $responseBuffer = true;
+
+    /**
+     * @param array{'@region'?: ?string, '@responseBuffer'?: bool, ...} $input
      */
     protected function __construct(array $input)
     {
         $this->region = $input['@region'] ?? null;
+        $this->responseBuffer = $input['@responseBuffer'] ?? true;
     }
 
     public function setRegion(?string $region): void
@@ -32,5 +38,10 @@ public function getRegion(): ?string
         return $this->region;
     }
 
+    public function shouldBufferResponse(): bool
+    {
+        return $this->responseBuffer;
+    }
+
     abstract public function request(): Request;
 }

src/Core/src/RequestContext.php

@@ -18,6 +18,7 @@ final class RequestContext
         'expirationDate' => true,
         'currentDate' => true,
         'exceptionMapping' => true,
+        'responseBuffer' => true,
         'usesEndpointDiscovery' => true,
         'requiresEndpointDiscovery' => true,
     ];
@@ -57,13 +58,19 @@ final class RequestContext
      */
     private $exceptionMapping = [];
 
+    /**
+     * @var bool
+     */
+    private $responseBuffer = true;
+
     /**
      * @param array{
      *  operation?: string|null,
      *  region?: string|null,
      *  expirationDate?: \DateTimeImmutable|null,
      *  currentDate?: \DateTimeImmutable|null,
      *  exceptionMapping?: array<string, class-string<HttpException>>,
+     *  responseBuffer?: bool,
      *  usesEndpointDiscovery?: bool,
      *  requiresEndpointDiscovery?: bool,
      * } $options
@@ -107,6 +114,11 @@ public function getExceptionMapping(): array
         return $this->exceptionMapping;
     }
 
+    public function shouldBufferResponse(): bool
+    {
+        return $this->responseBuffer;
+    }
+
     public function usesEndpointDiscovery(): bool
     {
         return $this->usesEndpointDiscovery;

src/Core/src/Response.php

@@ -17,6 +17,7 @@
 use AsyncAws\Core\Exception\LogicException;
 use AsyncAws\Core\Exception\RuntimeException;
 use AsyncAws\Core\Exception\UnparsableResponse;
+use AsyncAws\Core\Stream\ResponseBodyNonBufferedStream;
 use AsyncAws\Core\Stream\ResponseBodyStream;
 use AsyncAws\Core\Stream\ResultStream;
 use Psr\Log\LoggerInterface;
@@ -102,10 +103,15 @@ final class Response
      */
     private $exceptionMapping;
 
+    /**
+     * @var bool
+     */
+    private $responseBuffer;
+
     /**
      * @param array<string, class-string<HttpException>> $exceptionMapping
      */
-    public function __construct(ResponseInterface $response, HttpClientInterface $httpClient, LoggerInterface $logger, ?AwsErrorFactoryInterface $awsErrorFactory = null, ?EndpointCache $endpointCache = null, ?Request $request = null, bool $debug = false, array $exceptionMapping = [])
+    public function __construct(ResponseInterface $response, HttpClientInterface $httpClient, LoggerInterface $logger, ?AwsErrorFactoryInterface $awsErrorFactory = null, ?EndpointCache $endpointCache = null, ?Request $request = null, bool $debug = false, array $exceptionMapping = [], bool $responseBuffer = true)
     {
         $this->httpResponse = $response;
         $this->httpClient = $httpClient;
@@ -115,6 +121,7 @@ public function __construct(ResponseInterface $response, HttpClientInterface $ht
         $this->request = $request;
         $this->debug = $debug;
         $this->exceptionMapping = $exceptionMapping;
+        $this->responseBuffer = $responseBuffer;
     }
 
     public function __destruct()
@@ -166,12 +173,18 @@ public function resolve(?float $timeout = null): bool
                 // Network exception
                 $this->logger->debug('AsyncAws HTTP request could not be sent due network issues');
             } else {
+                if ($this->responseBuffer) {
+                    $body = $this->httpResponse->getContent(false);
+                    $this->bodyDownloaded = true;
+                } else {
+                    $body = '[response body omitted because buffering is disabled]';
+                }
+
                 $this->logger->debug('AsyncAws HTTP response received with status code {status_code}', [
                     'status_code' => $httpStatusCode,
                     'headers' => json_encode($this->httpResponse->getHeaders(false)),
-                    'body' => $this->httpResponse->getContent(false),
+                    'body' => $body,
                 ]);
-                $this->bodyDownloaded = true;
             }
         }
 
@@ -373,6 +386,15 @@ public function toStream(): ResultStream
         }
 
         try {
+            if (!$this->responseBuffer) {
+                $toStream = [$this->httpResponse, 'toStream'];
+                if (!\is_callable($toStream)) {
+                    throw new RuntimeException('The HTTP response does not support non-buffered streaming.');
+                }
+
+                return new ResponseBodyNonBufferedStream($toStream(false));
+            }
+
             return new ResponseBodyStream($this->httpClient->stream($this->httpResponse));
         } finally {
             $this->bodyDownloaded = true;