1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
<?php
namespace Aws\S3;
use GuzzleHttp\Promise\PromiseInterface;
use GuzzleHttp\Promise\PromisorInterface;
use GuzzleHttp\Psr7;
use Psr\Http\Message\StreamInterface;
/**
* Uploads an object to S3, using a PutObject command or a multipart upload as
* appropriate.
*/
class ObjectUploader implements PromisorInterface
{
const DEFAULT_MULTIPART_THRESHOLD = 16777216;
private $client;
private $bucket;
private $key;
private $body;
private $acl;
private $options;
private static $defaults = [
'before_upload' => null,
'concurrency' => 3,
'mup_threshold' => self::DEFAULT_MULTIPART_THRESHOLD,
'params' => [],
'part_size' => null,
];
/**
* @param S3ClientInterface $client The S3 Client used to execute
* the upload command(s).
* @param string $bucket Bucket to upload the object, or
* an S3 access point ARN.
* @param string $key Key of the object.
* @param mixed $body Object data to upload. Can be a
* StreamInterface, PHP stream
* resource, or a string of data to
* upload.
* @param string $acl ACL to apply to the copy
* (default: private).
* @param array $options Options used to configure the
* copy process. Options passed in
* through 'params' are added to
* the sub command(s).
*/
public function __construct(
S3ClientInterface $client,
$bucket,
$key,
$body,
$acl = 'private',
array $options = []
) {
$this->client = $client;
$this->bucket = $bucket;
$this->key = $key;
$this->body = Psr7\Utils::streamFor($body);
$this->acl = $acl;
$this->options = $options + self::$defaults;
}
/**
* @return PromiseInterface
*/
public function promise()
{
/** @var int $mup_threshold */
$mup_threshold = $this->options['mup_threshold'];
if ($this->requiresMultipart($this->body, $mup_threshold)) {
// Perform a multipart upload.
return (new MultipartUploader($this->client, $this->body, [
'bucket' => $this->bucket,
'key' => $this->key,
'acl' => $this->acl
] + $this->options))->promise();
}
// Perform a regular PutObject operation.
$command = $this->client->getCommand('PutObject', [
'Bucket' => $this->bucket,
'Key' => $this->key,
'Body' => $this->body,
'ACL' => $this->acl,
] + $this->options['params']);
if (is_callable($this->options['before_upload'])) {
$this->options['before_upload']($command);
}
return $this->client->executeAsync($command);
}
public function upload()
{
return $this->promise()->wait();
}
/**
* Determines if the body should be uploaded using PutObject or the
* Multipart Upload System. It also modifies the passed-in $body as needed
* to support the upload.
*
* @param StreamInterface $body Stream representing the body.
* @param integer $threshold Minimum bytes before using Multipart.
*
* @return bool
*/
private function requiresMultipart(StreamInterface &$body, $threshold)
{
// If body size known, compare to threshold to determine if Multipart.
if ($body->getSize() !== null) {
return $body->getSize() >= $threshold;
}
/**
* Handle the situation where the body size is unknown.
* Read up to 5MB into a buffer to determine how to upload the body.
* @var StreamInterface $buffer
*/
$buffer = Psr7\Utils::streamFor();
Psr7\Utils::copyToStream($body, $buffer, MultipartUploader::PART_MIN_SIZE);
// If body < 5MB, use PutObject with the buffer.
if ($buffer->getSize() < MultipartUploader::PART_MIN_SIZE) {
$buffer->seek(0);
$body = $buffer;
return false;
}
// If body >= 5 MB, then use multipart. [YES]
if ($body->isSeekable() && $body->getMetadata('uri') !== 'php://input') {
// If the body is seekable, just rewind the body.
$body->seek(0);
} else {
// If the body is non-seekable, stitch the rewind the buffer and
// the partially read body together into one stream. This avoids
// unnecessary disc usage and does not require seeking on the
// original stream.
$buffer->seek(0);
$body = new Psr7\AppendStream([$buffer, $body]);
}
return true;
}
}
|
