Viewing file:  ResumableUploader.php (8.64 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
<?php
/**
* Copyright 2016 Google Inc. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
namespace Google\Cloud\Core\Upload;
use Google\Cloud\Core\Exception\GoogleException;
use Google\Cloud\Core\Exception\ServiceException;
use Google\Cloud\Core\Exception\UploadException;
use Google\Cloud\Core\JsonTrait;
use Google\Cloud\Core\RequestWrapper;
use GuzzleHttp\Promise\PromiseInterface;
use GuzzleHttp\Psr7\LimitStream;
use GuzzleHttp\Psr7\Request;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\StreamInterface;
/**
* Resumable upload implementation.
*/
class ResumableUploader extends AbstractUploader
{
use JsonTrait;
/**
* @var callable
*/
private $uploadProgressCallback;
/**
* @var int
*/
protected $rangeStart = 0;
/**
* @var string
*/
protected $resumeUri;
/**
* Classes extending ResumableUploader may provide request headers to be
* included in {@see Google\Cloud\Core\Upload\ResumableUploader::upload()}
* and {@see Google\Cloud\Core\Upload\ResumableUploader::createResumeUri{}}.
*
* @var array
*/
protected $headers = [];
/**
* @param RequestWrapper $requestWrapper
* @param string|resource|StreamInterface $data
* @param string $uri
* @param array $options [optional] {
* Optional configuration.
*
* @type array $metadata Metadata on the resource.
* @type callable $uploadProgressCallback The given callable
* function/method will be called after each successfully uploaded
* chunk. The callable function/method will receive the number of
* uploaded bytes after each uploaded chunk as a parameter to this
* callable. It's useful if you want to create a progress bar when
* using resumable upload type together with $chunkSize parameter.
* If $chunkSize is not set the callable function/method will be
* called only once after the successful file upload.
* @type int $chunkSize Size of the chunks to send incrementally during
* a resumable upload. Must be in multiples of 262144 bytes.
* @type array $restOptions HTTP client specific configuration options.
* @type int $retries Number of retries for a failed request.
* **Defaults to** `3`.
* @type string $contentType Content type of the resource.
* }
*/
public function __construct(
RequestWrapper $requestWrapper,
$data,
$uri,
array $options = []
) {
parent::__construct($requestWrapper, $data, $uri, $options);
// Set uploadProgressCallback if it's passed as an option.
if (isset($options['uploadProgressCallback']) && is_callable($options['uploadProgressCallback'])) {
$this->uploadProgressCallback = $options['uploadProgressCallback'];
} elseif (isset($options['uploadProgressCallback'])) {
throw new \InvalidArgumentException('$options.uploadProgressCallback must be a callable.');
}
}
/**
* Gets the resume URI.
*
* @return string
*/
public function getResumeUri()
{
if (!$this->resumeUri) {
return $this->createResumeUri();
}
return $this->resumeUri;
}
/**
* Resumes a download using the provided URI.
*
* @param string $resumeUri
* @return array
* @throws GoogleException
*/
public function resume($resumeUri)
{
if (!$this->data->isSeekable()) {
throw new GoogleException('Cannot resume upload on a stream which cannot be seeked.');
}
$this->resumeUri = $resumeUri;
$response = $this->getStatusResponse();
if ($response->getBody()->getSize() > 0) {
return $this->decodeResponse($response);
}
$this->rangeStart = $this->getRangeStart($response->getHeaderLine('Range'));
return $this->upload();
}
/**
* Triggers the upload process.
*
* Errors are of form [`google.rpc.Status`](https://cloud.google.com/apis/design/errors#error_model),
* and may be obtained via {@see Google\Cloud\Core\Exception\ServiceException::getMetadata()}.
*
* @return array
* @throws ServiceException
*/
public function upload()
{
$rangeStart = $this->rangeStart;
$response = null;
$resumeUri = $this->getResumeUri();
$size = $this->data->getSize() ?: '*';
do {
$data = new LimitStream(
$this->data,
$this->chunkSize ?: - 1,
$rangeStart
);
$currStreamLimitSize = $data->getSize();
$rangeEnd = $rangeStart + ($currStreamLimitSize - 1);
$headers = $this->headers + [
'Content-Length' => $currStreamLimitSize,
'Content-Type' => $this->contentType,
'Content-Range' => "bytes $rangeStart-$rangeEnd/$size",
];
$request = new Request(
'PUT',
$resumeUri,
$headers,
$data
);
try {
$response = $this->requestWrapper->send($request, $this->requestOptions);
} catch (GoogleException $ex) {
throw new ServiceException(
"Upload failed. Please use this URI to resume your upload: $this->resumeUri",
$ex->getCode(),
null,
json_decode($ex->getMessage(), true) ?: []
);
}
if (is_callable($this->uploadProgressCallback)) {
call_user_func($this->uploadProgressCallback, $currStreamLimitSize);
}
$rangeStart = $this->getRangeStart($response->getHeaderLine('Range'));
} while ($response->getStatusCode() === 308);
return $this->decodeResponse($response);
}
/**
* Currently only the MultiPartUploader supports async.
*
* Any calls to this will throw a generic Google Exception.
*
* @return PromiseInterface
* @throws GoogleException
* @experimental The experimental flag means that while we believe this method
* or class is ready for use, it may change before release in backwards-
* incompatible ways. Please use with caution, and test thoroughly when
* upgrading.
*/
public function uploadAsync()
{
throw new GoogleException('Currently only the MultiPartUploader supports async.');
}
/**
* Fetch and decode the response body
*
* @param ResponseInterface $response
* @return array
*/
protected function decodeResponse(ResponseInterface $response)
{
return $this->jsonDecode($response->getBody(), true);
}
/**
* Creates the resume URI.
*
* @return string
*/
protected function createResumeUri()
{
$headers = $this->headers + [
'X-Upload-Content-Type' => $this->contentType,
'X-Upload-Content-Length' => $this->data->getSize(),
'Content-Type' => 'application/json'
];
$body = $this->jsonEncode($this->metadata);
$request = new Request(
'POST',
$this->uri,
$headers,
$body
);
$response = $this->requestWrapper->send($request, $this->requestOptions);
return $this->resumeUri = $response->getHeaderLine('Location');
}
/**
* Gets the status of the upload.
*
* @return ResponseInterface
*/
protected function getStatusResponse()
{
$request = new Request(
'PUT',
$this->resumeUri,
['Content-Range' => 'bytes */*']
);
return $this->requestWrapper->send($request, $this->requestOptions);
}
/**
* Gets the starting range for the upload.
*
* @param string $rangeHeader
* @return int|null
*/
protected function getRangeStart($rangeHeader)
{
if (!$rangeHeader) {
return null;
}
return (int) explode('-', $rangeHeader)[1] + 1;
}
}
|