123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203 |
- <?php
- namespace OC\Files\ObjectStore;
- use Aws\S3\Exception\S3MultipartUploadException;
- use Aws\S3\MultipartCopy;
- use Aws\S3\MultipartUploader;
- use Aws\S3\S3Client;
- use GuzzleHttp\Psr7;
- use GuzzleHttp\Psr7\Utils;
- use OC\Files\Stream\SeekableHttpStream;
- use Psr\Http\Message\StreamInterface;
- trait S3ObjectTrait {
- use S3ConfigTrait;
-
- abstract protected function getConnection();
- abstract protected function getCertificateBundlePath(): ?string;
- abstract protected function getSSECParameters(bool $copy = false): array;
-
- public function readObject($urn) {
- $fh = SeekableHttpStream::open(function ($range) use ($urn) {
- $command = $this->getConnection()->getCommand('GetObject', [
- 'Bucket' => $this->bucket,
- 'Key' => $urn,
- 'Range' => 'bytes=' . $range,
- ] + $this->getSSECParameters());
- $request = \Aws\serialize($command);
- $headers = [];
- foreach ($request->getHeaders() as $key => $values) {
- foreach ($values as $value) {
- $headers[] = "$key: $value";
- }
- }
- $opts = [
- 'http' => [
- 'protocol_version' => $request->getProtocolVersion(),
- 'header' => $headers,
- ]
- ];
- $bundle = $this->getCertificateBundlePath();
- if ($bundle) {
- $opts['ssl'] = [
- 'cafile' => $bundle
- ];
- }
- if ($this->getProxy()) {
- $opts['http']['proxy'] = $this->getProxy();
- $opts['http']['request_fulluri'] = true;
- }
- $context = stream_context_create($opts);
- return fopen($request->getUri(), 'r', false, $context);
- });
- if (!$fh) {
- throw new \Exception("Failed to read object $urn");
- }
- return $fh;
- }
-
- protected function writeSingle(string $urn, StreamInterface $stream, ?string $mimetype = null): void {
- $this->getConnection()->putObject([
- 'Bucket' => $this->bucket,
- 'Key' => $urn,
- 'Body' => $stream,
- 'ACL' => 'private',
- 'ContentType' => $mimetype,
- 'StorageClass' => $this->storageClass,
- ] + $this->getSSECParameters());
- }
-
- protected function writeMultiPart(string $urn, StreamInterface $stream, ?string $mimetype = null): void {
- $uploader = new MultipartUploader($this->getConnection(), $stream, [
- 'bucket' => $this->bucket,
- 'concurrency' => $this->concurrency,
- 'key' => $urn,
- 'part_size' => $this->uploadPartSize,
- 'params' => [
- 'ContentType' => $mimetype,
- 'StorageClass' => $this->storageClass,
- ] + $this->getSSECParameters(),
- ]);
- try {
- $uploader->upload();
- } catch (S3MultipartUploadException $e) {
-
-
- $uploadInfo = $e->getState()->getId();
- if ($e->getState()->isInitiated() && (array_key_exists('UploadId', $uploadInfo))) {
- $this->getConnection()->abortMultipartUpload($uploadInfo);
- }
- throw new \OCA\DAV\Connector\Sabre\Exception\BadGateway("Error while uploading to S3 bucket", 0, $e);
- }
- }
-
- public function writeObject($urn, $stream, ?string $mimetype = null) {
- $psrStream = Utils::streamFor($stream);
-
-
- $buffer = new Psr7\Stream(fopen("php://memory", 'rwb+'));
- Utils::copyToStream($psrStream, $buffer, $this->putSizeLimit);
- $buffer->seek(0);
- if ($buffer->getSize() < $this->putSizeLimit) {
-
- $this->writeSingle($urn, $buffer, $mimetype);
- } else {
- $loadStream = new Psr7\AppendStream([$buffer, $psrStream]);
- $this->writeMultiPart($urn, $loadStream, $mimetype);
- }
- }
-
- public function deleteObject($urn) {
- $this->getConnection()->deleteObject([
- 'Bucket' => $this->bucket,
- 'Key' => $urn,
- ]);
- }
- public function objectExists($urn) {
- return $this->getConnection()->doesObjectExist($this->bucket, $urn, $this->getSSECParameters());
- }
- public function copyObject($from, $to, array $options = []) {
- $sourceMetadata = $this->getConnection()->headObject([
- 'Bucket' => $this->getBucket(),
- 'Key' => $from,
- ] + $this->getSSECParameters());
- $size = (int)($sourceMetadata->get('Size') ?? $sourceMetadata->get('ContentLength'));
- if ($this->useMultipartCopy && $size > $this->copySizeLimit) {
- $copy = new MultipartCopy($this->getConnection(), [
- "source_bucket" => $this->getBucket(),
- "source_key" => $from
- ], array_merge([
- "bucket" => $this->getBucket(),
- "key" => $to,
- "acl" => "private",
- "params" => $this->getSSECParameters() + $this->getSSECParameters(true),
- "source_metadata" => $sourceMetadata
- ], $options));
- $copy->copy();
- } else {
- $this->getConnection()->copy($this->getBucket(), $from, $this->getBucket(), $to, 'private', array_merge([
- 'params' => $this->getSSECParameters() + $this->getSSECParameters(true),
- 'mup_threshold' => PHP_INT_MAX,
- ], $options));
- }
- }
- }
|