-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
AbstractUploadManager.php
321 lines (287 loc) · 10.3 KB
/
AbstractUploadManager.php
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
<?php
namespace Aws\Multipart;
use Aws\AwsClientInterface as Client;
use Aws\CommandInterface;
use Aws\CommandPool;
use Aws\Exception\AwsException;
use Aws\Exception\MultipartUploadException;
use Aws\Result;
use Aws\ResultInterface;
use GuzzleHttp\Promise;
use GuzzleHttp\Promise\PromiseInterface;
use InvalidArgumentException as IAE;
use Psr\Http\Message\RequestInterface;
/**
* Encapsulates the execution of a multipart upload to S3 or Glacier.
*
* @internal
*/
abstract class AbstractUploadManager implements Promise\PromisorInterface
{
const DEFAULT_CONCURRENCY = 5;
/** @var array Default values for base multipart configuration */
private static $defaultConfig = [
'part_size' => null,
'state' => null,
'concurrency' => self::DEFAULT_CONCURRENCY,
'prepare_data_source' => null,
'before_initiate' => null,
'before_upload' => null,
'before_complete' => null,
'exception_class' => 'Aws\Exception\MultipartUploadException',
];
/** @var Client Client used for the upload. */
protected $client;
/** @var array Configuration used to perform the upload. */
protected $config;
/** @var array Service-specific information about the upload workflow. */
protected $info;
/** @var PromiseInterface Promise that represents the multipart upload. */
protected $promise;
/** @var UploadState State used to manage the upload. */
protected $state;
/**
* @param Client $client
* @param array $config
*/
public function __construct(Client $client, array $config = [])
{
$this->client = $client;
$this->info = $this->loadUploadWorkflowInfo();
$this->config = $config + self::$defaultConfig;
$this->state = $this->determineState();
}
/**
* Returns the current state of the upload
*
* @return UploadState
*/
public function getState()
{
return $this->state;
}
/**
* Upload the source using multipart upload operations.
*
* @return Result The result of the CompleteMultipartUpload operation.
* @throws \LogicException if the upload is already complete or aborted.
* @throws MultipartUploadException if an upload operation fails.
*/
public function upload()
{
return $this->promise()->wait();
}
/**
* Upload the source asynchronously using multipart upload operations.
*
* @return PromiseInterface
*/
public function promise()
{
if ($this->promise) {
return $this->promise;
}
return $this->promise = Promise\coroutine(function () {
// Initiate the upload.
if ($this->state->isCompleted()) {
throw new \LogicException('This multipart upload has already '
. 'been completed or aborted.'
);
}
if (!$this->state->isInitiated()) {
// Execute the prepare callback.
if (is_callable($this->config["prepare_data_source"])) {
$this->config["prepare_data_source"]();
}
$result = (yield $this->execCommand('initiate', $this->getInitiateParams()));
$this->state->setUploadId(
$this->info['id']['upload_id'],
$result[$this->info['id']['upload_id']]
);
$this->state->setStatus(UploadState::INITIATED);
}
// Create a command pool from a generator that yields UploadPart
// commands for each upload part.
$resultHandler = $this->getResultHandler($errors);
$commands = new CommandPool(
$this->client,
$this->getUploadCommands($resultHandler),
[
'concurrency' => $this->config['concurrency'],
'before' => $this->config['before_upload'],
]
);
// Execute the pool of commands concurrently, and process errors.
yield $commands->promise();
if ($errors) {
throw new $this->config['exception_class']($this->state, $errors);
}
// Complete the multipart upload.
yield $this->execCommand('complete', $this->getCompleteParams());
$this->state->setStatus(UploadState::COMPLETED);
})->otherwise($this->buildFailureCatch());
}
private function transformException($e)
{
// Throw errors from the operations as a specific Multipart error.
if ($e instanceof AwsException) {
$e = new $this->config['exception_class']($this->state, $e);
}
throw $e;
}
private function buildFailureCatch()
{
if (interface_exists("Throwable")) {
return function (\Throwable $e) {
return $this->transformException($e);
};
} else {
return function (\Exception $e) {
return $this->transformException($e);
};
}
}
protected function getConfig()
{
return $this->config;
}
/**
* Provides service-specific information about the multipart upload
* workflow.
*
* This array of data should include the keys: 'command', 'id', and 'part_num'.
*
* @return array
*/
abstract protected function loadUploadWorkflowInfo();
/**
* Determines the part size to use for upload parts.
*
* Examines the provided partSize value and the source to determine the
* best possible part size.
*
* @throws \InvalidArgumentException if the part size is invalid.
*
* @return int
*/
abstract protected function determinePartSize();
/**
* Uses information from the Command and Result to determine which part was
* uploaded and mark it as uploaded in the upload's state.
*
* @param CommandInterface $command
* @param ResultInterface $result
*/
abstract protected function handleResult(
CommandInterface $command,
ResultInterface $result
);
/**
* Gets the service-specific parameters used to initiate the upload.
*
* @return array
*/
abstract protected function getInitiateParams();
/**
* Gets the service-specific parameters used to complete the upload.
*
* @return array
*/
abstract protected function getCompleteParams();
/**
* Based on the config and service-specific workflow info, creates a
* `Promise` for an `UploadState` object.
*
* @return PromiseInterface A `Promise` that resolves to an `UploadState`.
*/
private function determineState()
{
// If the state was provided via config, then just use it.
if ($this->config['state'] instanceof UploadState) {
return $this->config['state'];
}
// Otherwise, construct a new state from the provided identifiers.
$required = $this->info['id'];
$id = [$required['upload_id'] => null];
unset($required['upload_id']);
foreach ($required as $key => $param) {
if (!$this->config[$key]) {
throw new IAE('You must provide a value for "' . $key . '" in '
. 'your config for the MultipartUploader for '
. $this->client->getApi()->getServiceFullName() . '.');
}
$id[$param] = $this->config[$key];
}
$state = new UploadState($id);
$state->setPartSize($this->determinePartSize());
return $state;
}
/**
* Executes a MUP command with all of the parameters for the operation.
*
* @param string $operation Name of the operation.
* @param array $params Service-specific params for the operation.
*
* @return PromiseInterface
*/
protected function execCommand($operation, array $params)
{
// Create the command.
$command = $this->client->getCommand(
$this->info['command'][$operation],
$params + $this->state->getId()
);
// Execute the before callback.
if (is_callable($this->config["before_{$operation}"])) {
$this->config["before_{$operation}"]($command);
}
// Execute the command asynchronously and return the promise.
return $this->client->executeAsync($command);
}
/**
* Returns a middleware for processing responses of part upload operations.
*
* - Adds an onFulfilled callback that calls the service-specific
* handleResult method on the Result of the operation.
* - Adds an onRejected callback that adds the error to an array of errors.
* - Has a passedByRef $errors arg that the exceptions get added to. The
* caller should use that &$errors array to do error handling.
*
* @param array $errors Errors from upload operations are added to this.
*
* @return callable
*/
protected function getResultHandler(&$errors = [])
{
return function (callable $handler) use (&$errors) {
return function (
CommandInterface $command,
RequestInterface $request = null
) use ($handler, &$errors) {
return $handler($command, $request)->then(
function (ResultInterface $result) use ($command) {
$this->handleResult($command, $result);
return $result;
},
function (AwsException $e) use (&$errors) {
$errors[$e->getCommand()[$this->info['part_num']]] = $e;
return new Result();
}
);
};
};
}
/**
* Creates a generator that yields part data for the upload's source.
*
* Yields associative arrays of parameters that are ultimately merged in
* with others to form the complete parameters of a command. This can
* include the Body parameter, which is a limited stream (i.e., a Stream
* object, decorated with a LimitStream).
*
* @param callable $resultHandler
*
* @return \Generator
*/
abstract protected function getUploadCommands(callable $resultHandler);
}