s3Client = $s3Client; $this->requestArgs = $requestArgs; $this->validateConfig($config); $this->config = $config; $this->uploadId = $uploadId; $this->parts = $parts; $this->currentSnapshot = $currentSnapshot; $this->listenerNotifier = $listenerNotifier; } /** * @return PromiseInterface */ abstract protected function createMultipartOperation(): PromiseInterface; /** * @return PromiseInterface */ abstract protected function completeMultipartOperation(): PromiseInterface; /** * @return PromiseInterface */ abstract protected function processMultipartOperation(): PromiseInterface; /** * @return int */ abstract protected function getTotalSize(): int; /** * @param ResultInterface $result * * @return mixed */ abstract protected function createResponse(ResultInterface $result): mixed; /** * @param array $config * * @return void */ protected function validateConfig(array &$config): void { if (!isset($config['target_part_size_bytes'])) { $config['target_part_size_bytes'] = S3TransferManagerConfig::DEFAULT_TARGET_PART_SIZE_BYTES; } if (!isset($config['concurrency'])) { $config['concurrency'] = S3TransferManagerConfig::DEFAULT_CONCURRENCY; } $partSize = $config['target_part_size_bytes']; if ($partSize < self::PART_MIN_SIZE || $partSize > self::PART_MAX_SIZE) { throw new \InvalidArgumentException( "Part size config must be between " . self::PART_MIN_SIZE ." and " . self::PART_MAX_SIZE . " bytes " ."but it is configured to $partSize" ); } } /** * @return string|null */ public function getUploadId(): ?string { return $this->uploadId; } /** * @return array */ public function getParts(): array { return $this->parts; } /** * Get the current progress snapshot. * @return TransferProgressSnapshot|null */ public function getCurrentSnapshot(): ?TransferProgressSnapshot { return $this->currentSnapshot; } /** * @return PromiseInterface */ public function promise(): PromiseInterface { return Coroutine::of(function () { try { yield $this->createMultipartOperation(); yield $this->processMultipartOperation(); yield $this->completeMultipartOperation(); } finally { $this->callOnCompletionCallbacks(); } })->then(function (ResultInterface $result) { return $this->createResponse($result); })->otherwise(function (Throwable $e) { $this->operationFailed($e); throw $e; }); } /** * @return PromiseInterface */ protected function abortMultipartOperation(): PromiseInterface { $abortMultipartUploadArgs = $this->requestArgs; $abortMultipartUploadArgs['UploadId'] = $this->uploadId; $command = $this->s3Client->getCommand( 'AbortMultipartUpload', $abortMultipartUploadArgs ); return $this->s3Client->executeAsync($command); } /** * @return void */ protected function sortParts(): void { usort($this->parts, function ($partOne, $partTwo) { return $partOne['PartNumber'] <=> $partTwo['PartNumber']; }); } /** * @param ResultInterface $result * @param CommandInterface $command * @return void */ protected function collectPart( ResultInterface $result, CommandInterface $command ): void { $checksumResult = match($command->getName()) { 'UploadPart' => $result, 'UploadPartCopy' => $result['CopyPartResult'], default => $result[$command->getName() . 'Result'] }; $partData = [ 'PartNumber' => $command['PartNumber'], 'ETag' => $checksumResult['ETag'], ]; if (isset($command['ChecksumAlgorithm'])) { $checksumMemberName = 'Checksum' . strtoupper($command['ChecksumAlgorithm']); $partData[$checksumMemberName] = $checksumResult[$checksumMemberName] ?? null; } $this->parts[] = $partData; } /** * @param \Iterator $commands * @param callable $fulfilledCallback * @param callable $rejectedCallback * @return PromiseInterface */ protected function createCommandPool( \Iterator $commands, callable $fulfilledCallback, callable $rejectedCallback ): PromiseInterface { return (new CommandPool( $this->s3Client, $commands, [ 'concurrency' => $this->config['concurrency'], 'fulfilled' => $fulfilledCallback, 'rejected' => $rejectedCallback ] ))->promise(); } /** * @param array $requestArgs * @return void */ protected function operationInitiated(array $requestArgs): void { if ($this->currentSnapshot === null) { $this->currentSnapshot = new TransferProgressSnapshot( $requestArgs['Key'], 0, $this->getTotalSize() ); } $this->listenerNotifier?->transferInitiated([ AbstractTransferListener::REQUEST_ARGS_KEY => $requestArgs, AbstractTransferListener::PROGRESS_SNAPSHOT_KEY => $this->currentSnapshot ]); } /** * @param ResultInterface $result * @return void */ protected function operationCompleted(ResultInterface $result): void { $newSnapshot = new TransferProgressSnapshot( $this->currentSnapshot->getIdentifier(), $this->currentSnapshot->getTransferredBytes(), $this->currentSnapshot->getTotalBytes(), $result->toArray(), $this->currentSnapshot->getReason(), ); $this->currentSnapshot = $newSnapshot; $this->listenerNotifier?->transferComplete([ AbstractTransferListener::REQUEST_ARGS_KEY => $this->requestArgs, AbstractTransferListener::PROGRESS_SNAPSHOT_KEY => $this->currentSnapshot ]); } /** * @param Throwable $reason * @return void * */ protected function operationFailed(Throwable $reason): void { // Event already propagated if ($this->currentSnapshot?->getReason() !== null) { return; } if ($this->currentSnapshot === null) { $this->currentSnapshot = new TransferProgressSnapshot( 'Unknown', 0, 0, ); } $this->currentSnapshot = new TransferProgressSnapshot( $this->currentSnapshot->getIdentifier(), $this->currentSnapshot->getTransferredBytes(), $this->currentSnapshot->getTotalBytes(), $this->currentSnapshot->getResponse(), $reason ); if (!empty($this->uploadId)) { trigger_error( "Multipart Upload with id: " . $this->uploadId . " failed", ); $this->abortMultipartOperation()->wait(); } $this->listenerNotifier?->transferFail([ AbstractTransferListener::REQUEST_ARGS_KEY => $this->requestArgs, AbstractTransferListener::PROGRESS_SNAPSHOT_KEY => $this->currentSnapshot, 'reason' => $reason, ]); } /** * @param int $partSize * @param array $requestArgs * @return void */ protected function partCompleted( int $partSize, array $requestArgs ): void { $newSnapshot = new TransferProgressSnapshot( $this->currentSnapshot->getIdentifier(), $this->currentSnapshot->getTransferredBytes() + $partSize, $this->currentSnapshot->getTotalBytes(), $this->currentSnapshot->getResponse(), $this->currentSnapshot->getReason(), ); $this->currentSnapshot = $newSnapshot; $this->listenerNotifier?->bytesTransferred([ AbstractTransferListener::REQUEST_ARGS_KEY => $requestArgs, AbstractTransferListener::PROGRESS_SNAPSHOT_KEY => $this->currentSnapshot ]); } /** * @return void */ protected function callOnCompletionCallbacks(): void { foreach ($this->onCompletionCallbacks as $fn) { if (is_callable($fn)) { call_user_func($fn); } } $this->onCompletionCallbacks = []; } /** * @param Throwable $reason * @return void */ protected function partFailed(Throwable $reason): void { $this->operationFailed($reason); } /** * @return int */ protected function calculatePartSize(): int { return max( $this->getTotalSize() / self::PART_MAX_NUM, $this->config['target_part_size_bytes'] ); } }