media-cloud-sync / includes / sdk / s3 / Aws / Multipart / AbstractUploader.php

AbstractUploader.php in Media Cloud Sync trunk, at includes/sdk/s3/Aws/Multipart/AbstractUploader.php

121 lines 4.5 KB

Raw Download Zip

1	<?php
2
3	namespace Dudlewebs\WPMCS\s3\Aws\Multipart;
4
5	use Dudlewebs\WPMCS\s3\Aws\AwsClientInterface as Client;
6	use Dudlewebs\WPMCS\s3\Aws\Exception\AwsException;
7	use Dudlewebs\WPMCS\s3\GuzzleHttp\Psr7;
8	use InvalidArgumentException as IAE;
9	use Dudlewebs\WPMCS\s3\Psr\Http\Message\StreamInterface as Stream;
10	abstract class AbstractUploader extends AbstractUploadManager
11	{
12	/** @var Stream Source of the data to be uploaded. */
13	protected $source;
14	/** @var bool Configuration used to indicate if upload progress will be displayed. */
15	protected $displayProgress;
16	/**
17	* @param Client $client
18	* @param mixed $source
19	* @param array $config
20	*/
21	public function __construct(Client $client, $source, array $config = [])
22	{
23	$this->source = $this->determineSource($source);
24	parent::__construct($client, $config);
25	}
26	/**
27	* Create a stream for a part that starts at the current position and
28	* has a length of the upload part size (or less with the final part).
29	*
30	* @param Stream $stream
31	*
32	* @return Psr7\LimitStream
33	*/
34	protected function limitPartStream(Stream $stream)
35	{
36	// Limit what is read from the stream to the part size.
37	return new Psr7\LimitStream($stream, $this->state->getPartSize(), $this->source->tell());
38	}
39	protected function getUploadCommands(callable $resultHandler)
40	{
41	// Determine if the source can be seeked.
42	$seekable = $this->source->isSeekable() && $this->source->getMetadata('wrapper_type') === 'plainfile';
43	for ($partNumber = 1; $this->isEof($seekable); $partNumber++) {
44	// If we haven't already uploaded this part, yield a new part.
45	if (!$this->state->hasPartBeenUploaded($partNumber)) {
46	$partStartPos = $this->source->tell();
47	if (!($data = $this->createPart($seekable, $partNumber))) {
48	break;
49	}
50	$command = $this->client->getCommand($this->info['command']['upload'], $data + $this->state->getId());
51	$command->getHandlerList()->appendSign($resultHandler, 'mup');
52	$numberOfParts = $this->getNumberOfParts($this->state->getPartSize());
53	if (isset($numberOfParts) && $partNumber > $numberOfParts) {
54	throw new $this->config['exception_class']($this->state, new AwsException("Maximum part number for this job exceeded, file has likely been corrupted." . " Please restart this upload.", $command));
55	}
56	(yield $command);
57	if ($this->source->tell() > $partStartPos) {
58	continue;
59	}
60	}
61	// Advance the source's offset if not already advanced.
62	if ($seekable) {
63	$this->source->seek(\min($this->source->tell() + $this->state->getPartSize(), $this->source->getSize()));
64	} else {
65	$this->source->read($this->state->getPartSize());
66	}
67	}
68	}
69	/**
70	* Generates the parameters for an upload part by analyzing a range of the
71	* source starting from the current offset up to the part size.
72	*
73	* @param bool $seekable
74	* @param int $number
75	*
76	* @return array\|null
77	*/
78	protected abstract function createPart($seekable, $number);
79	/**
80	* Checks if the source is at EOF.
81	*
82	* @param bool $seekable
83	*
84	* @return bool
85	*/
86	private function isEof($seekable)
87	{
88	return $seekable ? $this->source->tell() < $this->source->getSize() : !$this->source->eof();
89	}
90	/**
91	* Turns the provided source into a stream and stores it.
92	*
93	* If a string is provided, it is assumed to be a filename, otherwise, it
94	* passes the value directly to `Psr7\Utils::streamFor()`.
95	*
96	* @param mixed $source
97	*
98	* @return Stream
99	*/
100	private function determineSource($source)
101	{
102	// Use the contents of a file as the data source.
103	if (\is_string($source)) {
104	$source = Psr7\Utils::tryFopen($source, 'r');
105	}
106	// Create a source stream.
107	$stream = Psr7\Utils::streamFor($source);
108	if (!$stream->isReadable()) {
109	throw new IAE('Source stream must be readable.');
110	}
111	return $stream;
112	}
113	protected function getNumberOfParts($partSize)
114	{
115	if ($sourceSize = $this->source->getSize()) {
116	return \ceil($sourceSize / $partSize);
117	}
118	return null;
119	}
120	}
121