| 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 |
|