Merge pull request #1797 from howardlopez/streaming_request_payloads

Streaming request payloads

.changes/nextrelease/streaming_request_payloads.json

@@ -0,0 +1,7 @@
+[
+    {
+        "type": "feature",
+        "category": "",
+        "description": "Adds support for 'requiresLength' trait, adding headers as necessary for streaming operations."
+    }
+]

src/AwsClient.php

@@ -184,6 +184,7 @@ public function __construct(array $args)
         $this->addClientSideMonitoring($args);
         $this->addEndpointParameterMiddleware($args);
         $this->addEndpointDiscoveryMiddleware($config, $args);
+        $this->addStreamRequestPayload();
 
         if (isset($args['with_resolved'])) {
             $args['with_resolved']($config);
@@ -371,6 +372,18 @@ private function addClientSideMonitoring($args)
         );
     }
 
+    private function addStreamRequestPayload()
+    {
+        $streamRequestPayloadMiddleware = StreamRequestPayloadMiddleware::wrap(
+            $this->api
+        );
+
+        $this->handlerList->prependSign(
+            $streamRequestPayloadMiddleware,
+            'StreamRequestPayloadMiddleware'
+        );
+    }
+
     /**
      * Returns a service model and doc model with any necessary changes
      * applied.

src/Exception/IncalculablePayloadException.php

@@ -0,0 +1,11 @@
+<?php
+namespace Aws\Exception;
+
+use Aws\HasMonitoringEventsTrait;
+use Aws\MonitoringEventsInterface;
+
+class IncalculablePayloadException extends \RuntimeException implements
+    MonitoringEventsInterface
+{
+    use HasMonitoringEventsTrait;
+}

src/StreamRequestPayloadMiddleware.php

@@ -0,0 +1,85 @@
+<?php
+namespace Aws;
+
+use Aws\Api\Service;
+use Aws\Exception\IncalculablePayloadException;
+use Psr\Http\Message\RequestInterface;
+
+/**
+ * @internal
+ */
+class StreamRequestPayloadMiddleware
+{
+    private $nextHandler;
+    private $service;
+
+    /**
+     * Create a middleware wrapper function
+     *
+     * @param Service $service
+     * @return \Closure
+     */
+    public static function wrap(Service $service)
+    {
+        return function (callable $handler) use ($service) {
+            return new self($handler, $service);
+        };
+    }
+
+    public function __construct(callable $nextHandler, Service $service)
+    {
+        $this->nextHandler = $nextHandler;
+        $this->service = $service;
+    }
+
+    public function __invoke(CommandInterface $command, RequestInterface $request)
+    {
+        $nextHandler = $this->nextHandler;
+
+        $operation = $this->service->getOperation($command->getName());
+        $contentLength = $request->getHeader('content-length');
+        $hasStreaming = false;
+        $requiresLength = false;
+
+        // Check if any present input member is a stream and requires the
+        // content length
+        foreach ($operation->getInput()->getMembers() as $name => $member) {
+            if (!empty($member['streaming']) && isset($command[$name])) {
+                $hasStreaming = true;
+                if (!empty($member['requiresLength'])) {
+                    $requiresLength = true;
+                }
+            }
+        }
+
+        if ($hasStreaming) {
+
+            // Add 'transfer-encoding' header if payload size not required to
+            // to be calculated and not already known
+            if (empty($requiresLength)
+                && empty($contentLength)
+                && isset($operation['authtype'])
+                && $operation['authtype'] == 'v4-unsigned-body'
+            ) {
+                $request = $request->withHeader('transfer-encoding', 'chunked');
+
+            // Otherwise, make sure 'content-length' header is added
+            } else {
+                if (empty($contentLength)) {
+                    $size = $request->getBody()->getSize();
+                    if (is_null($size)) {
+                        throw new IncalculablePayloadException('Payload'
+                            . ' content length is required and can not be'
+                            . ' calculated.');
+                    }
+                    $request = $request->withHeader(
+                        'content-length',
+                        $size
+                    );
+                }
+            }
+        }
+
+        return $nextHandler($command, $request);
+    }
+}