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

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

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