Struct qiniu_upload_manager::ObjectParams

impl ObjectParams

pub fn builder() -> ObjectParamsBuilder

创建对象上传参数构建器

pub fn region_provider(&self) -> Option<&dyn RegionsProvider>

获取区域信息提供者

Examples found in repository ?

src/single_part_uploader/form_uploader.rs (line 148)

    fn upload_path(&self, path: impl AsRef<Path>, params: ObjectParams) -> ApiResult<Value> {
        self.upload(
            params.region_provider(),
            self.make_request_body_from_path(path.as_ref(), self.make_upload_token_signer(&params).as_ref(), &params)?,
        )
    }

    fn upload_reader<R: Read + 'static>(&self, reader: R, params: ObjectParams) -> ApiResult<Value> {
        self.upload(
            params.region_provider(),
            self.make_request_body_from_reader(reader, None, self.make_upload_token_signer(&params).as_ref(), &params)?,
        )
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_upload_path<'a>(
        &'a self,
        path: impl AsRef<Path> + Send + Sync + 'a,
        params: ObjectParams,
    ) -> BoxFuture<'a, ApiResult<Value>> {
        Box::pin(async move {
            self.async_upload(
                params.region_provider(),
                self.make_async_request_body_from_path(
                    path.as_ref(),
                    self.make_upload_token_signer(&params).as_ref(),
                    &params,
                )
                .await?,
            )
            .await
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_upload_reader<R: AsyncRead + Unpin + Send + Sync + 'static>(
        &self,
        reader: R,
        params: ObjectParams,
    ) -> BoxFuture<ApiResult<Value>> {
        Box::pin(async move {
            self.async_upload(
                params.region_provider(),
                self.make_async_request_body_from_async_reader(
                    reader,
                    None,
                    self.make_upload_token_signer(&params).as_ref(),
                    &params,
                )
                .await?,
            )
            .await
        })
    }

More examples

Hide additional examples

src/multi_parts_uploader/mod.rs (line 321)

    fn get_bucket_regions(&self, params: &ObjectParams) -> ApiResult<GotRegions> {
        if let Some(region_provider) = params.region_provider() {
            region_provider.get_all(Default::default())
        } else {
            self.get_bucket_region()?.get_all(Default::default())
        }
    }

    #[cfg(feature = "async")]
    fn async_get_bucket_regions<'a>(&'a self, params: &'a ObjectParams) -> BoxFuture<'a, ApiResult<GotRegions>> {
        Box::pin(async move {
            if let Some(region_provider) = params.region_provider() {
                region_provider.async_get_all(Default::default()).await
            } else {
                self.async_get_bucket_region()
                    .await?
                    .async_get_all(Default::default())
                    .await
            }
        })
    }

    fn get_up_endpoints(&self, params: &ObjectParams) -> ApiResult<Endpoints> {
        let options = EndpointsGetOptions::builder().service_names(&[ServiceName::Up]).build();
        let up_endpoints = if let Some(region_provider) = params.region_provider() {
            RegionsProviderEndpoints::new(region_provider)
                .get_endpoints(options)?
                .into_owned()
        } else {
            RegionsProviderEndpoints::new(self.get_bucket_region()?)
                .get_endpoints(options)?
                .into_owned()
        };
        Ok(up_endpoints)
    }

    #[cfg(feature = "async")]
    fn async_get_up_endpoints<'a>(&'a self, params: &'a ObjectParams) -> BoxFuture<'a, ApiResult<Endpoints>> {
        Box::pin(async move {
            let options = EndpointsGetOptions::builder().service_names(&[ServiceName::Up]).build();
            let up_endpoints = if let Some(region_provider) = params.region_provider() {
                RegionsProviderEndpoints::new(region_provider)
                    .async_get_endpoints(options)
                    .await?
                    .into_owned()
            } else {
                RegionsProviderEndpoints::new(self.async_get_bucket_region().await?)
                    .async_get_endpoints(options)
                    .await?
                    .into_owned()
            };
            Ok(up_endpoints)
        })
    }

pub fn object_name(&self) -> Option<&str>

获取对象名称

Examples found in repository ?

src/single_part_uploader/form_uploader.rs (line 303)

    fn make_request_body_from_reader<'a, R: Read + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<SyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
            }
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = SyncRequestBody::default().set_upload_token(token, Default::default())?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_path<'a>(
        &'a self,
        path: &'a Path,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let file = AsyncFile::open(path).await?;
        self.make_async_request_body_from_async_reader(file, Some(path), token, params)
            .await
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_async_reader<'a, R: AsyncRead + Unpin + Send + Sync + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        let mut file_name_set = false;
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
            file_name_set = true;
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
                file_name_set = true;
            }
        }
        if !file_name_set {
            file_metadata = file_metadata.file_name("untitled");
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = AsyncRequestBody::default()
            .set_upload_token(token, Default::default())
            .await?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

    fn get_bucket_region(&self) -> ApiResult<BucketRegionsProvider> {
        Ok(self
            .upload_manager
            .queryer()
            .query(self.access_key()?, self.bucket_name()?))
    }

    #[cfg(feature = "async")]
    async fn async_get_bucket_region(&self) -> ApiResult<BucketRegionsProvider> {
        Ok(self
            .upload_manager
            .queryer()
            .query(self.async_access_key().await?, self.async_bucket_name().await?))
    }

    fn make_upload_token_signer(&self, params: &ObjectParams) -> OwnedUploadTokenProviderOrReferenced<'_> {
        let object_name = params.object_name().map(ObjectName::from);
        self.upload_manager
            .upload_token()
            .make_upload_token_provider(object_name)
    }

More examples

Hide additional examples

src/multi_parts_uploader/v1.rs (line 292)

    fn upload_part(
        &self,
        initialized: &Self::InitializedParts,
        data_partitioner_provider: &dyn DataPartitionProvider,
    ) -> ApiResult<Option<Self::UploadedPart>> {
        let data_partitioner_provider = normalize_data_partitioner_provider(data_partitioner_provider);
        let total_size = initialized.source.total_size()?;
        return if let Some(mut reader) = initialized.source.slice(data_partitioner_provider.part_size())? {
            if let Some(part_size) = NonZeroU64::new(reader.len()?) {
                let progresses_key = initialized.progresses.add_new_part(part_size.into());
                if let Some(uploaded_part) = _could_resume(initialized, &mut reader, part_size) {
                    self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                    Ok(Some(uploaded_part))
                } else {
                    let params = MkBlkPathParams::default().set_block_size_as_u64(part_size.into());
                    let upload_token_signer =
                        self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
                    let mkblk = self.storage().resumable_upload_v1_make_block();
                    let uploaded_result = _upload_part(
                        self,
                        mkblk.new_request(initialized.up_endpoints(), params, upload_token_signer.as_ref()),
                        reader,
                        part_size,
                        &progresses_key,
                        initialized,
                        &data_partitioner_provider,
                    );
                    match uploaded_result {
                        Ok(uploaded_part) => {
                            self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                            Ok(Some(uploaded_part))
                        }
                        Err(err) => {
                            self.after_part_uploaded(&progresses_key, total_size, None).ok();
                            Err(err)
                        }
                    }
                }
            } else {
                Ok(None)
            }
        } else {
            Ok(None)
        };

        fn _could_resume<H: Digest>(
            initialized: &MultiPartsV1UploaderInitializedObject<Box<dyn DataSource<H>>>,
            data_reader: &mut DataSourceReader,
            part_size: NonZeroU64,
        ) -> Option<MultiPartsV1UploaderUploadedPart> {
            let offset = data_reader.offset();
            initialized.resumed_records.take(offset).and_then(|record| {
                if record.size == part_size
                    && record.expired_at() > SystemTime::now()
                    && Some(record.base64ed_sha1.as_str()) == sha1_of_sync_reader(data_reader).ok().as_deref()
                {
                    Some(MultiPartsV1UploaderUploadedPart {
                        response_body: record.response_body.to_owned(),
                        uploaded_size: record.size,
                        resumed: true,
                        offset,
                    })
                } else {
                    None
                }
            })
        }

        #[allow(clippy::too_many_arguments)]
        fn _upload_part<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV1Uploader<H>,
            mut request: SyncMkBlkRequestBuilder<'a, E>,
            mut body: DataSourceReader,
            content_length: NonZeroU64,
            progresses_key: &'a ProgressesKey,
            initialized: &'a MultiPartsV1UploaderInitializedObject<Box<dyn DataSource<H>>>,
            data_partitioner_provider: &'a dyn DataPartitionProvider,
        ) -> ApiResult<MultiPartsV1UploaderUploadedPart> {
            let total_size = initialized.source.total_size()?;
            request.on_uploading_progress(move |_, transfer| {
                progresses_key.update_part(transfer.transferred_bytes());
                uploader.callbacks.upload_progress(&UploadingProgressInfo::new(
                    progresses_key.current_uploaded(),
                    total_size,
                ))
            });
            uploader.before_request_call(request.parts_mut())?;
            let base64ed_sha1 = sha1_of_sync_reader(&mut body)?;
            let body_offset = body.offset();
            let begin_at = Instant::now();
            let mut response_result = request.call(body, content_length.get());
            let elapsed = begin_at.elapsed();
            uploader.after_response_call(&mut response_result)?;
            let mut feedback_builder =
                DataPartitionProviderFeedback::builder(content_length.into(), elapsed, initialized.params.extensions());
            if let Some(err) = response_result.as_ref().err() {
                feedback_builder.error(err);
            }
            data_partitioner_provider.feedback(feedback_builder.build());
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let response_body = response_result?.into_body();
            let record = MultiPartsV1ResumableRecorderRecord {
                expired_timestamp: response_body.get_expired_at_as_u64(),
                offset: body_offset,
                size: content_length,
                base64ed_sha1,
                response_body,
            };
            initialized
                .resumed_records
                .persist(&record, &uploader.bucket_name()?, initialized.up_endpoints())
                .ok();
            Ok(MultiPartsV1UploaderUploadedPart::from_record(record, false))
        }
    }

    fn complete_parts(&self, initialized: &Self::InitializedParts, parts: &[Self::UploadedPart]) -> ApiResult<Value> {
        let file_size = get_file_size_from_uploaded_parts(parts);
        let upload_token_signer = self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
        let params = make_mkfile_path_params_from_initialized_parts(&initialized.params, file_size);
        let mkfile = self.storage().resumable_upload_v1_make_file();
        let body = make_mkfile_request_body_from_uploaded_parts(parts.to_vec());
        return _complete_parts(
            self,
            mkfile.new_request(initialized.up_endpoints(), params, upload_token_signer.as_ref()),
            &initialized.source,
            body,
        );

        fn _complete_parts<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a, D: DataSource<H>>(
            uploader: &'a MultiPartsV1Uploader<H>,
            mut request: SyncMkFileRequestBuilder<'a, E>,
            source: &D,
            body: String,
        ) -> ApiResult<Value> {
            uploader.before_request_call(request.parts_mut())?;
            let content_length = body.len() as u64;
            let mut response_result = request.call(Cursor::new(body), content_length);
            uploader.after_response_call(&mut response_result)?;
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let body = response_result?.into_body();
            uploader.try_to_delete_records(&source).ok();
            Ok(body.into())
        }
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    type AsyncInitializedParts = MultiPartsV1UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>;

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    type AsyncUploadedPart = MultiPartsV1UploaderUploadedPart;

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_initialize_parts<D: AsyncDataSource<Self::HashAlgorithm> + 'static>(
        &self,
        source: D,
        params: ObjectParams,
    ) -> BoxFuture<ApiResult<Self::AsyncInitializedParts>> {
        Box::pin(async move {
            let up_endpoints = self.async_get_up_endpoints(&params).await?;
            let resumed_records = self.async_resume_or_create_records(&source, up_endpoints).await?;
            Ok(Self::AsyncInitializedParts {
                source: Box::new(source),
                params,
                resumed_records,
                progresses: Default::default(),
            })
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn try_to_async_resume_parts<D: AsyncDataSource<Self::HashAlgorithm> + 'static>(
        &self,
        source: D,
        params: ObjectParams,
    ) -> BoxFuture<Option<Self::AsyncInitializedParts>> {
        Box::pin(async move {
            if let Some(source_key) = source.source_key().await.ok().flatten() {
                self.try_to_async_resume_records(&source_key)
                    .await
                    .ok()
                    .flatten()
                    .map(|resumed_records| Self::AsyncInitializedParts {
                        source: Box::new(source),
                        params,
                        resumed_records,
                        progresses: Default::default(),
                    })
            } else {
                None
            }
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_reinitialize_parts<'r>(
        &'r self,
        initialized: &'r mut Self::AsyncInitializedParts,
        options: ReinitializeOptions,
    ) -> BoxFuture<'r, ApiResult<()>> {
        Box::pin(async move {
            initialized.source.reset().await?;
            let up_endpoints = options.async_get_up_endpoints(self, initialized).await?;
            initialized.resumed_records = self.async_create_new_records(&initialized.source, up_endpoints).await?;
            initialized.progresses = Default::default();
            Ok(())
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_upload_part<'r>(
        &'r self,
        initialized: &'r Self::AsyncInitializedParts,
        data_partitioner_provider: &'r dyn DataPartitionProvider,
    ) -> BoxFuture<'r, ApiResult<Option<Self::AsyncUploadedPart>>> {
        return Box::pin(async move {
            let data_partitioner_provider = normalize_data_partitioner_provider(data_partitioner_provider);
            let total_size = initialized.source.total_size().await?;
            if let Some(mut reader) = initialized.source.slice(data_partitioner_provider.part_size()).await? {
                if let Some(part_size) = NonZeroU64::new(reader.len().await?) {
                    let progresses_key = initialized.progresses.add_new_part(part_size.into());
                    if let Some(uploaded_part) = _could_resume(initialized, &mut reader, part_size).await {
                        self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                        Ok(Some(uploaded_part))
                    } else {
                        let params = MkBlkPathParams::default().set_block_size_as_u64(part_size.into());
                        let upload_token_signer =
                            self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
                        let mkblk = self.storage().resumable_upload_v1_make_block();
                        let uploaded_result = _upload_part(
                            self,
                            mkblk.new_async_request(initialized.up_endpoints(), params, upload_token_signer.as_ref()),
                            reader,
                            part_size,
                            &progresses_key,
                            initialized,
                            &data_partitioner_provider,
                        )
                        .await;
                        match uploaded_result {
                            Ok(uploaded_part) => {
                                self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                                Ok(Some(uploaded_part))
                            }
                            Err(err) => {
                                self.after_part_uploaded(&progresses_key, total_size, None).ok();
                                Err(err)
                            }
                        }
                    }
                } else {
                    Ok(None)
                }
            } else {
                Ok(None)
            }
        });

        async fn _could_resume<H: Digest>(
            initialized: &MultiPartsV1UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>,
            data_reader: &mut AsyncDataSourceReader,
            part_size: NonZeroU64,
        ) -> Option<MultiPartsV1UploaderUploadedPart> {
            let offset = data_reader.offset();
            OptionFuture::from(initialized.resumed_records.take(offset).map(|record| async move {
                if record.size == part_size
                    && record.expired_at() > SystemTime::now()
                    && Some(record.base64ed_sha1.as_str()) == sha1_of_async_reader(data_reader).await.ok().as_deref()
                {
                    Some(MultiPartsV1UploaderUploadedPart {
                        response_body: record.response_body.to_owned(),
                        uploaded_size: record.size,
                        resumed: true,
                        offset,
                    })
                } else {
                    None
                }
            }))
            .await
            .flatten()
        }

        #[allow(clippy::too_many_arguments)]
        async fn _upload_part<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV1Uploader<H>,
            mut request: AsyncMkBlkRequestBuilder<'a, E>,
            mut body: AsyncDataSourceReader,
            content_length: NonZeroU64,
            progresses_key: &'a ProgressesKey,
            initialized: &'a MultiPartsV1UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>,
            data_partitioner_provider: &'a dyn DataPartitionProvider,
        ) -> ApiResult<MultiPartsV1UploaderUploadedPart> {
            let total_size = initialized.source.total_size().await?;
            request.on_uploading_progress(move |_, transfer| {
                progresses_key.update_part(transfer.transferred_bytes());
                uploader.callbacks.upload_progress(&UploadingProgressInfo::new(
                    progresses_key.current_uploaded(),
                    total_size,
                ))
            });
            uploader.before_request_call(request.parts_mut())?;
            let body_offset = body.offset();
            let base64ed_sha1 = sha1_of_async_reader(&mut body).await?;
            let begin_at = Instant::now();
            let mut response_result = request.call(body, content_length.get()).await;
            let elapsed = begin_at.elapsed();
            uploader.after_response_call(&mut response_result)?;
            let mut feedback_builder =
                DataPartitionProviderFeedback::builder(content_length.into(), elapsed, initialized.params.extensions());
            if let Some(err) = response_result.as_ref().err() {
                feedback_builder.error(err);
            }
            data_partitioner_provider.feedback(feedback_builder.build());
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let response_body = response_result?.into_body();
            let record = MultiPartsV1ResumableRecorderRecord {
                expired_timestamp: response_body.get_expired_at_as_u64(),
                offset: body_offset,
                size: content_length,
                base64ed_sha1,
                response_body,
            };
            initialized
                .resumed_records
                .async_persist(
                    &record,
                    &uploader.async_bucket_name().await?,
                    initialized.up_endpoints(),
                )
                .await
                .ok();
            Ok(MultiPartsV1UploaderUploadedPart::from_record(record, false))
        }
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_complete_parts<'r>(
        &'r self,
        initialized: &'r Self::AsyncInitializedParts,
        parts: &'r [Self::AsyncUploadedPart],
    ) -> BoxFuture<'r, ApiResult<Value>> {
        return Box::pin(async move {
            let file_size = get_file_size_from_uploaded_parts(parts);
            let upload_token_signer = self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
            let params = make_mkfile_path_params_from_initialized_parts(&initialized.params, file_size);
            let mkfile = self.storage().resumable_upload_v1_make_file();
            let body = make_mkfile_request_body_from_uploaded_parts(parts.to_vec());
            _complete_parts(
                self,
                mkfile.new_async_request(initialized.up_endpoints(), params, upload_token_signer.as_ref()),
                &initialized.source,
                body,
            )
            .await
        });

        async fn _complete_parts<
            'a,
            H: Digest + Send + 'static,
            E: EndpointsProvider + Clone + 'a,
            D: AsyncDataSource<H>,
        >(
            uploader: &'a MultiPartsV1Uploader<H>,
            mut request: AsyncMkFileRequestBuilder<'a, E>,
            source: &D,
            body: String,
        ) -> ApiResult<Value> {
            uploader.before_request_call(request.parts_mut())?;
            let content_length = body.len() as u64;
            let mut response_result = request.call(AsyncCursor::new(body), content_length).await;
            uploader.after_response_call(&mut response_result)?;
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let body = response_result?.into_body();
            uploader.try_to_async_delete_records(source).await.ok();
            Ok(body.into())
        }
    }
}

impl<H: Digest> super::__private::Sealed for MultiPartsV1Uploader<H> {}

fn make_mkfile_path_params_from_initialized_parts(object_params: &ObjectParams, file_size: u64) -> MkFilePathParams {
    let mut params = MkFilePathParams::default().set_size_as_u64(file_size);
    if let Some(object_name) = object_params.object_name() {
        params = params.set_object_name_as_str(object_name.to_string());
    }
    if let Some(file_name) = object_params.file_name() {
        params = params.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = object_params.content_type() {
        params = params.set_mime_type_as_str(mime.to_string());
    }
    for (metadata_name, metadata_value) in object_params.metadata() {
        params = params.append_custom_data_as_str("x-qn-meta-".to_owned() + metadata_name, metadata_value.to_owned());
    }
    for (var_name, var_value) in object_params.custom_vars() {
        params = params.append_custom_data_as_str("x:".to_owned() + var_name, var_value.to_owned());
    }
    params
}

src/multi_parts_uploader/v2.rs (line 343)

    fn upload_part(
        &self,
        initialized: &Self::InitializedParts,
        data_partitioner_provider: &dyn DataPartitionProvider,
    ) -> ApiResult<Option<Self::UploadedPart>> {
        let data_partitioner_provider = normalize_data_partitioner_provider(data_partitioner_provider);
        let total_size = initialized.source.total_size()?;
        return if let Some(mut reader) = initialized.source.slice(data_partitioner_provider.part_size())? {
            if let Some(part_size) = NonZeroU64::new(reader.len()?) {
                let progresses_key = initialized.progresses.add_new_part(part_size.into());
                if let Some(uploaded_part) = _could_resume(initialized, &mut reader, part_size) {
                    self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                    Ok(Some(uploaded_part))
                } else {
                    let path_params = make_upload_part_path_params_from_initialized_params(
                        self.bucket_name()?.to_string(),
                        &initialized.params,
                        initialized.info.upload_id.to_owned(),
                        reader.part_number(),
                    );
                    let upload_token_signer =
                        self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
                    let upload_part = self.storage().resumable_upload_v2_upload_part();
                    let uploaded_result = _upload_part(
                        self,
                        upload_part.new_request(&initialized.up_endpoints(), path_params, upload_token_signer.as_ref()),
                        reader,
                        part_size,
                        initialized,
                        &progresses_key,
                        &data_partitioner_provider,
                    );

                    match uploaded_result {
                        Ok(uploaded_part) => {
                            self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                            Ok(Some(uploaded_part))
                        }
                        Err(err) => {
                            self.after_part_uploaded(&progresses_key, total_size, None).ok();
                            Err(err)
                        }
                    }
                }
            } else {
                Ok(None)
            }
        } else {
            Ok(None)
        };

        fn _could_resume<H: Digest>(
            initialized: &MultiPartsV2UploaderInitializedObject<Box<dyn DataSource<H>>>,
            data_reader: &mut DataSourceReader,
            part_size: NonZeroU64,
        ) -> Option<MultiPartsV2UploaderUploadedPart> {
            let offset = data_reader.offset();
            initialized.resumed_records.take(offset).and_then(|record| {
                if record.size == part_size
                    && record.part_number == data_reader.part_number()
                    && Some(record.base64ed_sha1.as_str()) == sha1_of_sync_reader(data_reader).ok().as_deref()
                {
                    Some(MultiPartsV2UploaderUploadedPart {
                        response_body: record.response_body.to_owned(),
                        uploaded_size: record.size,
                        part_number: record.part_number,
                        resumed: true,
                        offset,
                    })
                } else {
                    None
                }
            })
        }

        fn _upload_part<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: SyncUploadPartRequestBuilder<'a, E>,
            mut body: DataSourceReader,
            content_length: NonZeroU64,
            initialized: &'a MultiPartsV2UploaderInitializedObject<Box<dyn DataSource<H>>>,
            progresses_key: &'a ProgressesKey,
            data_partitioner_provider: &'a dyn DataPartitionProvider,
        ) -> ApiResult<MultiPartsV2UploaderUploadedPart> {
            let total_size = initialized.source.total_size()?;
            request.on_uploading_progress(move |_, transfer| {
                progresses_key.update_part(transfer.transferred_bytes());
                uploader.callbacks.upload_progress(&UploadingProgressInfo::new(
                    progresses_key.current_uploaded(),
                    total_size,
                ))
            });
            uploader.before_request_call(request.parts_mut())?;
            let body_offset = body.offset();
            let part_number = body.part_number();
            let base64ed_sha1 = sha1_of_sync_reader(&mut body)?;
            let begin_at = Instant::now();
            let mut response_result = request.call(body, content_length.get());
            let elapsed = begin_at.elapsed();
            uploader.after_response_call(&mut response_result)?;
            let mut feedback_builder =
                DataPartitionProviderFeedback::builder(content_length.into(), elapsed, initialized.params.extensions());
            if let Some(err) = response_result.as_ref().err() {
                feedback_builder.error(err);
            }
            data_partitioner_provider.feedback(feedback_builder.build());
            let response_body = response_result?.into_body();
            let record = MultiPartsV2ResumableRecorderRecord {
                response_body,
                offset: body_offset,
                size: content_length,
                base64ed_sha1,
                part_number,
            };
            initialized
                .resumed_records
                .persist(
                    &initialized.info,
                    &record,
                    &uploader.bucket_name()?,
                    initialized.params.object_name(),
                    initialized.up_endpoints(),
                )
                .ok();
            Ok(MultiPartsV2UploaderUploadedPart::from_record(record, false))
        }
    }

    fn complete_parts(&self, initialized: &Self::InitializedParts, parts: &[Self::UploadedPart]) -> ApiResult<Value> {
        let upload_token_signer = self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
        let path_params = make_complete_parts_path_params_from_initialized_params(
            self.bucket_name()?.to_string(),
            &initialized.params,
            initialized.info.upload_id.to_owned(),
        );
        let body = make_complete_parts_request_body_from_initialized_params(&initialized.params, parts.to_vec());
        let complete_parts = self.storage().resumable_upload_v2_complete_multipart_upload();
        return _complete_parts(
            self,
            complete_parts.new_request(&initialized.up_endpoints(), path_params, upload_token_signer.as_ref()),
            &initialized.source,
            body,
        );

        fn _complete_parts<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a, D: DataSource<H>>(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: SyncCompletePartsRequestBuilder<'a, E>,
            source: &D,
            body: CompletePartsRequestBody,
        ) -> ApiResult<Value> {
            uploader.before_request_call(request.parts_mut())?;
            let mut response_result = request.call(&body);
            uploader.after_response_call(&mut response_result)?;
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let body = response_result?.into_body();
            uploader.try_to_delete_records(&source).ok();
            Ok(body.into())
        }
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    type AsyncInitializedParts = MultiPartsV2UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>;

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    type AsyncUploadedPart = MultiPartsV2UploaderUploadedPart;

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_initialize_parts<D: AsyncDataSource<Self::HashAlgorithm> + 'static>(
        &self,
        source: D,
        params: ObjectParams,
    ) -> BoxFuture<ApiResult<Self::AsyncInitializedParts>> {
        Box::pin(async move {
            let (info, resumed_records) = self
                .async_resume_or_create_records(&source, &params, self.async_get_up_endpoints(&params).await?)
                .await?;
            let info = if let Some(info) = info {
                info
            } else {
                self.async_call_initialize_parts(&params, &resumed_records.up_endpoints)
                    .await?
            };

            Ok(Self::AsyncInitializedParts {
                source: Box::new(source),
                params,
                info,
                resumed_records,
                progresses: Default::default(),
            })
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn try_to_async_resume_parts<D: AsyncDataSource<Self::HashAlgorithm> + 'static>(
        &self,
        source: D,
        params: ObjectParams,
    ) -> BoxFuture<Option<Self::AsyncInitializedParts>> {
        Box::pin(async move {
            if let Some(source_key) = source.source_key().await.ok().flatten() {
                self.try_to_async_resume_records(&source_key, &params)
                    .await
                    .ok()
                    .flatten()
                    .map(|(info, resumed_records)| Self::AsyncInitializedParts {
                        source: Box::new(source),
                        params,
                        info,
                        resumed_records,
                        progresses: Default::default(),
                    })
            } else {
                None
            }
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_reinitialize_parts<'r>(
        &'r self,
        initialized: &'r mut Self::AsyncInitializedParts,
        options: ReinitializeOptions,
    ) -> BoxFuture<ApiResult<()>> {
        Box::pin(async move {
            initialized.source.reset().await?;
            let up_endpoints = options.async_get_up_endpoints(self, initialized).await?;
            initialized.resumed_records = self.async_create_new_records(&initialized.source, up_endpoints).await?;
            initialized.info = self
                .async_call_initialize_parts(&initialized.params, initialized.up_endpoints())
                .await?;
            initialized.progresses = Default::default();
            Ok(())
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_upload_part<'r>(
        &'r self,
        initialized: &'r Self::AsyncInitializedParts,
        data_partitioner_provider: &'r dyn DataPartitionProvider,
    ) -> BoxFuture<'r, ApiResult<Option<Self::AsyncUploadedPart>>> {
        return Box::pin(async move {
            let data_partitioner_provider = normalize_data_partitioner_provider(data_partitioner_provider);
            let total_size = initialized.source.total_size().await?;
            if let Some(mut reader) = initialized.source.slice(data_partitioner_provider.part_size()).await? {
                if let Some(part_size) = NonZeroU64::new(reader.len().await?) {
                    let progresses_key = initialized.progresses.add_new_part(part_size.get());
                    if let Some(uploaded_part) = _could_resume(initialized, &mut reader, part_size).await {
                        self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                        Ok(Some(uploaded_part))
                    } else {
                        let path_params = make_upload_part_path_params_from_initialized_params(
                            self.bucket_name()?.to_string(),
                            &initialized.params,
                            initialized.info.upload_id.to_owned(),
                            reader.part_number(),
                        );
                        let upload_token_signer =
                            self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
                        let upload_part = self.storage().resumable_upload_v2_upload_part();
                        let uploaded_result = _upload_part(
                            self,
                            upload_part.new_async_request(
                                initialized.up_endpoints(),
                                path_params,
                                upload_token_signer.as_ref(),
                            ),
                            reader,
                            part_size,
                            initialized,
                            &progresses_key,
                            &data_partitioner_provider,
                        )
                        .await;
                        match uploaded_result {
                            Ok(uploaded_part) => {
                                self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                                Ok(Some(uploaded_part))
                            }
                            Err(err) => {
                                self.after_part_uploaded(&progresses_key, total_size, None).ok();
                                Err(err)
                            }
                        }
                    }
                } else {
                    Ok(None)
                }
            } else {
                Ok(None)
            }
        });

        async fn _could_resume<H: Digest>(
            initialized: &MultiPartsV2UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>,
            data_reader: &mut AsyncDataSourceReader,
            part_size: NonZeroU64,
        ) -> Option<MultiPartsV2UploaderUploadedPart> {
            let offset = data_reader.offset();
            OptionFuture::from(initialized.resumed_records.take(offset).map(|record| async move {
                if record.size == part_size
                    && record.part_number == data_reader.part_number()
                    && Some(record.base64ed_sha1.as_str()) == sha1_of_async_reader(data_reader).await.ok().as_deref()
                {
                    Some(MultiPartsV2UploaderUploadedPart {
                        response_body: record.response_body.to_owned(),
                        uploaded_size: record.size,
                        part_number: record.part_number,
                        resumed: true,
                        offset,
                    })
                } else {
                    None
                }
            }))
            .await
            .flatten()
        }

        async fn _upload_part<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: AsyncUploadPartRequestBuilder<'a, E>,
            mut body: AsyncDataSourceReader,
            content_length: NonZeroU64,
            initialized: &'a MultiPartsV2UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>,
            progresses_key: &'a ProgressesKey,
            data_partitioner_provider: &'a dyn DataPartitionProvider,
        ) -> ApiResult<MultiPartsV2UploaderUploadedPart> {
            let total_size = initialized.source.total_size().await?;
            request.on_uploading_progress(move |_, transfer| {
                progresses_key.update_part(transfer.transferred_bytes());
                uploader.callbacks.upload_progress(&UploadingProgressInfo::new(
                    progresses_key.current_uploaded(),
                    total_size,
                ))
            });
            uploader.before_request_call(request.parts_mut())?;
            let body_offset = body.offset();
            let part_number = body.part_number();
            let base64ed_sha1 = sha1_of_async_reader(&mut body).await?;
            let begin_at = Instant::now();
            let mut response_result = request.call(body, content_length.get()).await;
            let elapsed = begin_at.elapsed();
            uploader.after_response_call(&mut response_result)?;
            let mut feedback_builder =
                DataPartitionProviderFeedback::builder(content_length.into(), elapsed, initialized.params.extensions());
            if let Some(err) = response_result.as_ref().err() {
                feedback_builder.error(err);
            }
            data_partitioner_provider.feedback(feedback_builder.build());
            let response_body = response_result?.into_body();
            let record = MultiPartsV2ResumableRecorderRecord {
                response_body,
                offset: body_offset,
                size: content_length,
                base64ed_sha1,
                part_number,
            };
            initialized
                .resumed_records
                .async_persist(
                    &initialized.info,
                    &record,
                    &uploader.async_bucket_name().await?,
                    initialized.params.object_name(),
                    initialized.up_endpoints(),
                )
                .await
                .ok();
            Ok(MultiPartsV2UploaderUploadedPart::from_record(record, false))
        }
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_complete_parts<'r>(
        &'r self,
        initialized: &'r Self::AsyncInitializedParts,
        parts: &'r [Self::AsyncUploadedPart],
    ) -> BoxFuture<'r, ApiResult<Value>> {
        return Box::pin(async move {
            let upload_token_signer = self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
            let path_params = make_complete_parts_path_params_from_initialized_params(
                self.bucket_name()?.to_string(),
                &initialized.params,
                initialized.info.upload_id.to_owned(),
            );
            let body = make_complete_parts_request_body_from_initialized_params(&initialized.params, parts.to_vec());
            let complete_parts = self.storage().resumable_upload_v2_complete_multipart_upload();
            _complete_parts(
                self,
                complete_parts.new_async_request(initialized.up_endpoints(), path_params, upload_token_signer.as_ref()),
                &initialized.source,
                body,
            )
            .await
        });

        async fn _complete_parts<
            'a,
            H: Digest + Send + 'static,
            E: EndpointsProvider + Clone + 'a,
            D: AsyncDataSource<H>,
        >(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: AsyncCompletePartsRequestBuilder<'a, E>,
            source: &D,
            body: CompletePartsRequestBody,
        ) -> ApiResult<Value> {
            uploader.before_request_call(request.parts_mut())?;
            let mut response_result = request.call(&body).await;
            uploader.after_response_call(&mut response_result)?;
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let body = response_result?.into_body();
            uploader.try_to_async_delete_records(&source).await.ok();
            Ok(body.into())
        }
    }
}

impl<H: Digest> super::__private::Sealed for MultiPartsV2Uploader<H> {}

fn make_init_parts_path_params_from_initialized_params(
    bucket_name: String,
    params: &ObjectParams,
) -> InitPartsPathParams {
    let mut path_params = InitPartsPathParams::default().set_bucket_name_as_str(bucket_name);
    if let Some(object_name) = params.object_name() {
        path_params = path_params.set_object_name_as_str(object_name.to_owned());
    }
    path_params
}

fn make_upload_part_path_params_from_initialized_params(
    bucket_name: String,
    params: &ObjectParams,
    upload_id: String,
    part_number: NonZeroUsize,
) -> UploadPartPathParams {
    let mut path_params = UploadPartPathParams::default()
        .set_bucket_name_as_str(bucket_name)
        .set_upload_id_as_str(upload_id)
        .set_part_number_as_usize(part_number.get());
    if let Some(object_name) = params.object_name() {
        path_params = path_params.set_object_name_as_str(object_name.to_owned());
    }
    path_params
}

fn make_complete_parts_path_params_from_initialized_params(
    bucket_name: String,
    params: &ObjectParams,
    upload_id: String,
) -> CompletePartsPathParams {
    let mut path_params = CompletePartsPathParams::default()
        .set_bucket_name_as_str(bucket_name)
        .set_upload_id_as_str(upload_id);
    if let Some(object_name) = params.object_name() {
        path_params = path_params.set_object_name_as_str(object_name.to_owned());
    }
    path_params
}

fn make_complete_parts_request_body_from_initialized_params(
    params: &ObjectParams,
    mut parts: Vec<MultiPartsV2UploaderUploadedPart>,
) -> CompletePartsRequestBody {
    parts.sort_by_key(|part| part.part_number);
    let mut body = CompletePartsRequestBody::default();
    body.set_parts(
        parts
            .iter()
            .map(|part| {
                let mut part_info = PartInfo::default();
                part_info.set_etag_as_str(part.response_body.get_etag_as_str().to_owned());
                part_info.set_part_number_as_u64(part.part_number.get() as u64);
                part_info
            })
            .collect::<Vec<_>>()
            .into(),
    );
    if let Some(file_name) = params.file_name() {
        body.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = params.content_type() {
        body.set_mime_type_as_str(mime.to_string());
    }
    body.set_metadata(StringMap::from(
        params
            .metadata()
            .iter()
            .map(|(key, value)| ("x-qn-meta-".to_owned() + key, value.to_owned())),
    ));
    body.set_custom_vars(StringMap::from(
        params
            .custom_vars()
            .iter()
            .map(|(key, value)| ("x:".to_owned() + key, value.to_owned())),
    ));
    body
}

fn sha1_of_sync_reader<R: Read + Reset>(reader: &mut R) -> IoResult<String> {
    Ok(urlsafe_base64(Digestible::<Sha1>::digest(reader)?.as_slice()))
}

#[cfg(feature = "async")]
async fn sha1_of_async_reader<R: AsyncRead + AsyncReset + Unpin + Send>(reader: &mut R) -> IoResult<String> {
    Ok(urlsafe_base64(
        AsyncDigestible::<Sha1>::digest(reader).await?.as_slice(),
    ))
}

fn may_set_extensions_in_err(err: &mut ResponseError) {
    match err.kind() {
        ResponseErrorKind::StatusCodeError(status_code) if status_code.as_u16() == 612 => {
            err.extensions_mut().insert(PartsExpiredError);
        }
        _ => {}
    }
}

fn normalize_data_partitioner_provider<P: DataPartitionProvider>(base: P) -> LimitedDataPartitionProvider<P> {
    LimitedDataPartitionProvider::new_with_non_zero_threshold(base, MIN_PART_SIZE, MAX_PART_SIZE)
}

impl<H: Digest + Send + 'static> MultiPartsV2Uploader<H> {
    fn before_request_call(&self, request: &mut RequestBuilderParts<'_>) -> ApiResult<()> {
        self.callbacks.before_request(request).map_err(make_callback_error)
    }

    fn after_response_call<B>(&self, response: &mut ApiResult<Response<B>>) -> ApiResult<()> {
        self.callbacks.after_response(response).map_err(make_callback_error)
    }

    fn after_part_uploaded(
        &self,
        progresses_key: &ProgressesKey,
        total_size: Option<u64>,
        uploaded_part: Option<&MultiPartsV2UploaderUploadedPart>,
    ) -> ApiResult<()> {
        if let Some(uploaded_part) = uploaded_part {
            progresses_key.complete_part();
            self.callbacks
                .part_uploaded(uploaded_part)
                .map_err(make_callback_error)?;
        } else {
            progresses_key.delete_part();
        }
        self.callbacks
            .upload_progress(&UploadingProgressInfo::new(
                progresses_key.current_uploaded(),
                total_size,
            ))
            .map_err(make_callback_error)
    }

    fn call_initialize_parts(
        &self,
        params: &ObjectParams,
        up_endpoints: &Endpoints,
    ) -> ApiResult<MultiPartsV2UploaderInitializedResponseInfo> {
        let path_params = make_init_parts_path_params_from_initialized_params(self.bucket_name()?.to_string(), params);
        let upload_token_signer = self.make_upload_token_signer(params.object_name().map(|n| n.into()));
        let init_parts = self.storage().resumable_upload_v2_initiate_multipart_upload();
        let body = _initialize_parts(
            self,
            init_parts.new_request(&up_endpoints, path_params, upload_token_signer.as_ref()),
        )?;
        return Ok(MultiPartsV2UploaderInitializedResponseInfo {
            upload_id: body.get_upload_id_as_str().to_owned(),
            expired_at: UNIX_EPOCH + Duration::from_secs(body.get_expired_at_as_u64()),
        });

        fn _initialize_parts<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: SyncInitPartsRequestBuilder<'a, E>,
        ) -> ApiResult<InitPartsResponseBody> {
            uploader.before_request_call(request.parts_mut())?;
            let mut response_result = request.call();
            uploader.after_response_call(&mut response_result)?;
            Ok(response_result?.into_body())
        }
    }

    #[cfg(feature = "async")]
    async fn async_call_initialize_parts(
        &self,
        params: &ObjectParams,
        up_endpoints: &Endpoints,
    ) -> ApiResult<MultiPartsV2UploaderInitializedResponseInfo> {
        let path_params =
            make_init_parts_path_params_from_initialized_params(self.async_bucket_name().await?.to_string(), params);
        let upload_token_signer = self.make_upload_token_signer(params.object_name().map(|n| n.into()));
        let init_parts = self.storage().resumable_upload_v2_initiate_multipart_upload();
        let initialized_parts = _initialize_parts(
            self,
            init_parts.new_async_request(up_endpoints, path_params, upload_token_signer.as_ref()),
        )
        .await?;
        return Ok(MultiPartsV2UploaderInitializedResponseInfo {
            upload_id: initialized_parts.get_upload_id_as_str().to_owned(),
            expired_at: UNIX_EPOCH + Duration::from_secs(initialized_parts.get_expired_at_as_u64()),
        });

        async fn _initialize_parts<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: AsyncInitPartsRequestBuilder<'a, E>,
        ) -> ApiResult<InitPartsResponseBody> {
            uploader.before_request_call(request.parts_mut())?;
            let mut response_result = request.call().await;
            uploader.after_response_call(&mut response_result)?;
            Ok(response_result?.into_body())
        }
    }

    fn resume_or_create_records<D: DataSource<H>>(
        &self,
        source: &D,
        params: &ObjectParams,
        up_endpoints: Endpoints,
    ) -> IoResult<(
        Option<MultiPartsV2UploaderInitializedResponseInfo>,
        MultiPartsV2ResumableRecorderRecords,
    )> {
        let (info, records) = if let Some(source_key) = source.source_key()? {
            self.try_to_resume_records(&source_key, params)
                .ok()
                .flatten()
                .map_or_else(
                    || {
                        (
                            None,
                            _create_new_records(&self.resumable_recorder, &source_key, up_endpoints),
                        )
                    },
                    |(info, records)| (Some(info), records),
                )
        } else {
            (None, MultiPartsV2ResumableRecorderRecords::new(up_endpoints))
        };
        Ok((info, records))
    }

    fn try_to_resume_records(
        &self,
        source_key: &SourceKey<H>,
        params: &ObjectParams,
    ) -> ApiResult<
        Option<(
            MultiPartsV2UploaderInitializedResponseInfo,
            MultiPartsV2ResumableRecorderRecords,
        )>,
    > {
        let mut medium = self.resumable_recorder.open_for_read(source_key)?;
        let mut lines = BufReader::new(&mut medium).lines();
        let (info, up_endpoints) = if let Some(line) = lines.next() {
            let line = line?;
            let mut header: MultiPartsV2ResumableRecorderDeserializableHeader = serde_json::from_str(&line)?;
            if !header.is_v2()
                || header.expired_at() <= SystemTime::now()
                || header.bucket() != self.bucket_name()?.as_str()
                || header.object() != params.object_name()
            {
                return Ok(None);
            }
            let info = MultiPartsV2UploaderInitializedResponseInfo {
                upload_id: take(&mut header.upload_id),
                expired_at: header.expired_at(),
            };
            (info, header.up_endpoints)
        } else {
            return Ok(None);
        };
        let mut records = lines
            .map(|line| {
                let line = line?;
                let record: MultiPartsV2ResumableRecorderRecord = serde_json::from_str(&line)?;
                Ok(record)
            })
            .collect::<ApiResult<MultiPartsV2ResumableRecorderRecords>>()?;
        records.up_endpoints = up_endpoints;
        records.set_medium_for_append(self.resumable_recorder.open_for_append(source_key)?, true);
        Ok(Some((info, records)))
    }

    fn create_new_records<D: DataSource<H>>(
        &self,
        source: &D,
        up_endpoints: Endpoints,
    ) -> IoResult<MultiPartsV2ResumableRecorderRecords> {
        let records = if let Some(source_key) = source.source_key()? {
            _create_new_records(&self.resumable_recorder, &source_key, up_endpoints)
        } else {
            MultiPartsV2ResumableRecorderRecords::new(up_endpoints)
        };
        Ok(records)
    }

    #[cfg(feature = "async")]
    async fn async_resume_or_create_records<D: AsyncDataSource<H>>(
        &self,
        source: &D,
        params: &ObjectParams,
        up_endpoints: Endpoints,
    ) -> IoResult<(
        Option<MultiPartsV2UploaderInitializedResponseInfo>,
        MultiPartsV2ResumableRecorderRecords,
    )> {
        let result = if let Some(source_key) = source.source_key().await? {
            if let Some((info, records)) = self
                .try_to_async_resume_records(&source_key, params)
                .await
                .ok()
                .flatten()
            {
                (Some(info), records)
            } else {
                (
                    None,
                    _async_create_new_records(&self.resumable_recorder, &source_key, up_endpoints).await,
                )
            }
        } else {
            (None, MultiPartsV2ResumableRecorderRecords::new(up_endpoints))
        };
        Ok(result)
    }

    #[cfg(feature = "async")]
    async fn try_to_async_resume_records(
        &self,
        source_key: &SourceKey<H>,
        params: &ObjectParams,
    ) -> ApiResult<
        Option<(
            MultiPartsV2UploaderInitializedResponseInfo,
            MultiPartsV2ResumableRecorderRecords,
        )>,
    > {
        let mut medium = self.resumable_recorder.open_for_async_read(source_key).await?;
        let mut lines = AsyncBufReader::new(&mut medium).lines();
        let (info, up_endpoints) = if let Some(line) = lines.try_next().await? {
            let mut header: MultiPartsV2ResumableRecorderDeserializableHeader = serde_json::from_str(&line)?;
            if !header.is_v2()
                || header.expired_at() <= SystemTime::now()
                || header.bucket() != self.bucket_name()?.as_str()
                || header.object() != params.object_name()
            {
                return Ok(None);
            }
            let info = MultiPartsV2UploaderInitializedResponseInfo {
                upload_id: take(&mut header.upload_id),
                expired_at: header.expired_at(),
            };
            (info, header.up_endpoints)
        } else {
            return Ok(None);
        };
        let mut records = lines
            .map(|line| {
                let line = line?;
                let record: MultiPartsV2ResumableRecorderRecord = serde_json::from_str(&line)?;
                Ok::<_, ResponseError>(record)
            })
            .try_collect::<MultiPartsV2ResumableRecorderRecords>()
            .await?;
        records.up_endpoints = up_endpoints;
        records.set_medium_for_async_append(self.resumable_recorder.open_for_async_append(source_key).await?, true);
        Ok(Some((info, records)))
    }

pub fn file_name(&self) -> Option<&str>

获取文件名称

Examples found in repository ?

src/multi_parts_uploader/v1.rs (line 678)

fn make_mkfile_path_params_from_initialized_parts(object_params: &ObjectParams, file_size: u64) -> MkFilePathParams {
    let mut params = MkFilePathParams::default().set_size_as_u64(file_size);
    if let Some(object_name) = object_params.object_name() {
        params = params.set_object_name_as_str(object_name.to_string());
    }
    if let Some(file_name) = object_params.file_name() {
        params = params.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = object_params.content_type() {
        params = params.set_mime_type_as_str(mime.to_string());
    }
    for (metadata_name, metadata_value) in object_params.metadata() {
        params = params.append_custom_data_as_str("x-qn-meta-".to_owned() + metadata_name, metadata_value.to_owned());
    }
    for (var_name, var_value) in object_params.custom_vars() {
        params = params.append_custom_data_as_str("x:".to_owned() + var_name, var_value.to_owned());
    }
    params
}

More examples

Hide additional examples

src/multi_parts_uploader/v2.rs (line 814)

fn make_complete_parts_request_body_from_initialized_params(
    params: &ObjectParams,
    mut parts: Vec<MultiPartsV2UploaderUploadedPart>,
) -> CompletePartsRequestBody {
    parts.sort_by_key(|part| part.part_number);
    let mut body = CompletePartsRequestBody::default();
    body.set_parts(
        parts
            .iter()
            .map(|part| {
                let mut part_info = PartInfo::default();
                part_info.set_etag_as_str(part.response_body.get_etag_as_str().to_owned());
                part_info.set_part_number_as_u64(part.part_number.get() as u64);
                part_info
            })
            .collect::<Vec<_>>()
            .into(),
    );
    if let Some(file_name) = params.file_name() {
        body.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = params.content_type() {
        body.set_mime_type_as_str(mime.to_string());
    }
    body.set_metadata(StringMap::from(
        params
            .metadata()
            .iter()
            .map(|(key, value)| ("x-qn-meta-".to_owned() + key, value.to_owned())),
    ));
    body.set_custom_vars(StringMap::from(
        params
            .custom_vars()
            .iter()
            .map(|(key, value)| ("x:".to_owned() + key, value.to_owned())),
    ));
    body
}

src/single_part_uploader/form_uploader.rs (line 292)

    fn make_request_body_from_reader<'a, R: Read + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<SyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
            }
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = SyncRequestBody::default().set_upload_token(token, Default::default())?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_path<'a>(
        &'a self,
        path: &'a Path,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let file = AsyncFile::open(path).await?;
        self.make_async_request_body_from_async_reader(file, Some(path), token, params)
            .await
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_async_reader<'a, R: AsyncRead + Unpin + Send + Sync + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        let mut file_name_set = false;
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
            file_name_set = true;
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
                file_name_set = true;
            }
        }
        if !file_name_set {
            file_metadata = file_metadata.file_name("untitled");
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = AsyncRequestBody::default()
            .set_upload_token(token, Default::default())
            .await?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

pub fn content_type(&self) -> Option<&Mime>

获取 MIME 类型

Examples found in repository ?

src/multi_parts_uploader/v1.rs (line 681)

fn make_mkfile_path_params_from_initialized_parts(object_params: &ObjectParams, file_size: u64) -> MkFilePathParams {
    let mut params = MkFilePathParams::default().set_size_as_u64(file_size);
    if let Some(object_name) = object_params.object_name() {
        params = params.set_object_name_as_str(object_name.to_string());
    }
    if let Some(file_name) = object_params.file_name() {
        params = params.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = object_params.content_type() {
        params = params.set_mime_type_as_str(mime.to_string());
    }
    for (metadata_name, metadata_value) in object_params.metadata() {
        params = params.append_custom_data_as_str("x-qn-meta-".to_owned() + metadata_name, metadata_value.to_owned());
    }
    for (var_name, var_value) in object_params.custom_vars() {
        params = params.append_custom_data_as_str("x:".to_owned() + var_name, var_value.to_owned());
    }
    params
}

More examples

Hide additional examples

src/multi_parts_uploader/v2.rs (line 817)

fn make_complete_parts_request_body_from_initialized_params(
    params: &ObjectParams,
    mut parts: Vec<MultiPartsV2UploaderUploadedPart>,
) -> CompletePartsRequestBody {
    parts.sort_by_key(|part| part.part_number);
    let mut body = CompletePartsRequestBody::default();
    body.set_parts(
        parts
            .iter()
            .map(|part| {
                let mut part_info = PartInfo::default();
                part_info.set_etag_as_str(part.response_body.get_etag_as_str().to_owned());
                part_info.set_part_number_as_u64(part.part_number.get() as u64);
                part_info
            })
            .collect::<Vec<_>>()
            .into(),
    );
    if let Some(file_name) = params.file_name() {
        body.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = params.content_type() {
        body.set_mime_type_as_str(mime.to_string());
    }
    body.set_metadata(StringMap::from(
        params
            .metadata()
            .iter()
            .map(|(key, value)| ("x-qn-meta-".to_owned() + key, value.to_owned())),
    ));
    body.set_custom_vars(StringMap::from(
        params
            .custom_vars()
            .iter()
            .map(|(key, value)| ("x:".to_owned() + key, value.to_owned())),
    ));
    body
}

src/single_part_uploader/form_uploader.rs (line 299)

    fn make_request_body_from_reader<'a, R: Read + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<SyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
            }
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = SyncRequestBody::default().set_upload_token(token, Default::default())?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_path<'a>(
        &'a self,
        path: &'a Path,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let file = AsyncFile::open(path).await?;
        self.make_async_request_body_from_async_reader(file, Some(path), token, params)
            .await
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_async_reader<'a, R: AsyncRead + Unpin + Send + Sync + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        let mut file_name_set = false;
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
            file_name_set = true;
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
                file_name_set = true;
            }
        }
        if !file_name_set {
            file_metadata = file_metadata.file_name("untitled");
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = AsyncRequestBody::default()
            .set_upload_token(token, Default::default())
            .await?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

pub fn metadata(&self) -> &HashMap<String, String>

获取对象元信息

Examples found in repository ?

src/multi_parts_uploader/v1.rs (line 684)

fn make_mkfile_path_params_from_initialized_parts(object_params: &ObjectParams, file_size: u64) -> MkFilePathParams {
    let mut params = MkFilePathParams::default().set_size_as_u64(file_size);
    if let Some(object_name) = object_params.object_name() {
        params = params.set_object_name_as_str(object_name.to_string());
    }
    if let Some(file_name) = object_params.file_name() {
        params = params.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = object_params.content_type() {
        params = params.set_mime_type_as_str(mime.to_string());
    }
    for (metadata_name, metadata_value) in object_params.metadata() {
        params = params.append_custom_data_as_str("x-qn-meta-".to_owned() + metadata_name, metadata_value.to_owned());
    }
    for (var_name, var_value) in object_params.custom_vars() {
        params = params.append_custom_data_as_str("x:".to_owned() + var_name, var_value.to_owned());
    }
    params
}

More examples

Hide additional examples

src/multi_parts_uploader/v2.rs (line 822)

fn make_complete_parts_request_body_from_initialized_params(
    params: &ObjectParams,
    mut parts: Vec<MultiPartsV2UploaderUploadedPart>,
) -> CompletePartsRequestBody {
    parts.sort_by_key(|part| part.part_number);
    let mut body = CompletePartsRequestBody::default();
    body.set_parts(
        parts
            .iter()
            .map(|part| {
                let mut part_info = PartInfo::default();
                part_info.set_etag_as_str(part.response_body.get_etag_as_str().to_owned());
                part_info.set_part_number_as_u64(part.part_number.get() as u64);
                part_info
            })
            .collect::<Vec<_>>()
            .into(),
    );
    if let Some(file_name) = params.file_name() {
        body.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = params.content_type() {
        body.set_mime_type_as_str(mime.to_string());
    }
    body.set_metadata(StringMap::from(
        params
            .metadata()
            .iter()
            .map(|(key, value)| ("x-qn-meta-".to_owned() + key, value.to_owned())),
    ));
    body.set_custom_vars(StringMap::from(
        params
            .custom_vars()
            .iter()
            .map(|(key, value)| ("x:".to_owned() + key, value.to_owned())),
    ));
    body
}

src/single_part_uploader/form_uploader.rs (line 306)

    fn make_request_body_from_reader<'a, R: Read + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<SyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
            }
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = SyncRequestBody::default().set_upload_token(token, Default::default())?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_path<'a>(
        &'a self,
        path: &'a Path,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let file = AsyncFile::open(path).await?;
        self.make_async_request_body_from_async_reader(file, Some(path), token, params)
            .await
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_async_reader<'a, R: AsyncRead + Unpin + Send + Sync + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        let mut file_name_set = false;
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
            file_name_set = true;
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
                file_name_set = true;
            }
        }
        if !file_name_set {
            file_metadata = file_metadata.file_name("untitled");
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = AsyncRequestBody::default()
            .set_upload_token(token, Default::default())
            .await?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

pub fn custom_vars(&self) -> &HashMap<String, String>

获取对象自定义变量

Examples found in repository ?

src/multi_parts_uploader/v1.rs (line 687)

fn make_mkfile_path_params_from_initialized_parts(object_params: &ObjectParams, file_size: u64) -> MkFilePathParams {
    let mut params = MkFilePathParams::default().set_size_as_u64(file_size);
    if let Some(object_name) = object_params.object_name() {
        params = params.set_object_name_as_str(object_name.to_string());
    }
    if let Some(file_name) = object_params.file_name() {
        params = params.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = object_params.content_type() {
        params = params.set_mime_type_as_str(mime.to_string());
    }
    for (metadata_name, metadata_value) in object_params.metadata() {
        params = params.append_custom_data_as_str("x-qn-meta-".to_owned() + metadata_name, metadata_value.to_owned());
    }
    for (var_name, var_value) in object_params.custom_vars() {
        params = params.append_custom_data_as_str("x:".to_owned() + var_name, var_value.to_owned());
    }
    params
}

More examples

Hide additional examples

src/multi_parts_uploader/v2.rs (line 828)

fn make_complete_parts_request_body_from_initialized_params(
    params: &ObjectParams,
    mut parts: Vec<MultiPartsV2UploaderUploadedPart>,
) -> CompletePartsRequestBody {
    parts.sort_by_key(|part| part.part_number);
    let mut body = CompletePartsRequestBody::default();
    body.set_parts(
        parts
            .iter()
            .map(|part| {
                let mut part_info = PartInfo::default();
                part_info.set_etag_as_str(part.response_body.get_etag_as_str().to_owned());
                part_info.set_part_number_as_u64(part.part_number.get() as u64);
                part_info
            })
            .collect::<Vec<_>>()
            .into(),
    );
    if let Some(file_name) = params.file_name() {
        body.set_file_name_as_str(file_name.to_string());
    }
    if let Some(mime) = params.content_type() {
        body.set_mime_type_as_str(mime.to_string());
    }
    body.set_metadata(StringMap::from(
        params
            .metadata()
            .iter()
            .map(|(key, value)| ("x-qn-meta-".to_owned() + key, value.to_owned())),
    ));
    body.set_custom_vars(StringMap::from(
        params
            .custom_vars()
            .iter()
            .map(|(key, value)| ("x:".to_owned() + key, value.to_owned())),
    ));
    body
}

src/single_part_uploader/form_uploader.rs (line 309)

    fn make_request_body_from_reader<'a, R: Read + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<SyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
            }
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = SyncRequestBody::default().set_upload_token(token, Default::default())?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_path<'a>(
        &'a self,
        path: &'a Path,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let file = AsyncFile::open(path).await?;
        self.make_async_request_body_from_async_reader(file, Some(path), token, params)
            .await
    }

    #[cfg(feature = "async")]
    async fn make_async_request_body_from_async_reader<'a, R: AsyncRead + Unpin + Send + Sync + 'static>(
        &'a self,
        reader: R,
        path: Option<&'a Path>,
        token: &'a (dyn UploadTokenProvider + 'a),
        params: &'a ObjectParams,
    ) -> ApiResult<AsyncRequestBody<'a>> {
        let mut file_metadata = PartMetadata::default();
        let mut file_name_set = false;
        if let Some(file_name) = params.file_name() {
            file_metadata = file_metadata.file_name(file_name);
            file_name_set = true;
        } else if let Some(path) = path {
            if let Some(file_name) = path.file_name() {
                file_metadata = file_metadata.file_name(Path::new(file_name).display().to_string());
                file_name_set = true;
            }
        }
        if !file_name_set {
            file_metadata = file_metadata.file_name("untitled");
        }
        if let Some(content_type) = params.content_type() {
            file_metadata = file_metadata.mime(content_type.to_owned());
        }
        let mut request_body = AsyncRequestBody::default()
            .set_upload_token(token, Default::default())
            .await?;
        if let Some(object_name) = params.object_name() {
            request_body = request_body.set_object_name(object_name.to_string());
        }
        for (key, value) in params.metadata() {
            request_body = request_body.append_custom_data("x-qn-meta-".to_owned() + key, value);
        }
        for (key, value) in params.custom_vars() {
            request_body = request_body.append_custom_data("x:".to_owned() + key, value);
        }
        request_body = request_body.set_file_as_reader(reader, file_metadata);
        Ok(request_body)
    }

pub fn extensions(&self) -> &Extensions

获取扩展信息

Examples found in repository ?

src/multi_parts_uploader/v1.rs (line 370)

        fn _upload_part<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV1Uploader<H>,
            mut request: SyncMkBlkRequestBuilder<'a, E>,
            mut body: DataSourceReader,
            content_length: NonZeroU64,
            progresses_key: &'a ProgressesKey,
            initialized: &'a MultiPartsV1UploaderInitializedObject<Box<dyn DataSource<H>>>,
            data_partitioner_provider: &'a dyn DataPartitionProvider,
        ) -> ApiResult<MultiPartsV1UploaderUploadedPart> {
            let total_size = initialized.source.total_size()?;
            request.on_uploading_progress(move |_, transfer| {
                progresses_key.update_part(transfer.transferred_bytes());
                uploader.callbacks.upload_progress(&UploadingProgressInfo::new(
                    progresses_key.current_uploaded(),
                    total_size,
                ))
            });
            uploader.before_request_call(request.parts_mut())?;
            let base64ed_sha1 = sha1_of_sync_reader(&mut body)?;
            let body_offset = body.offset();
            let begin_at = Instant::now();
            let mut response_result = request.call(body, content_length.get());
            let elapsed = begin_at.elapsed();
            uploader.after_response_call(&mut response_result)?;
            let mut feedback_builder =
                DataPartitionProviderFeedback::builder(content_length.into(), elapsed, initialized.params.extensions());
            if let Some(err) = response_result.as_ref().err() {
                feedback_builder.error(err);
            }
            data_partitioner_provider.feedback(feedback_builder.build());
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let response_body = response_result?.into_body();
            let record = MultiPartsV1ResumableRecorderRecord {
                expired_timestamp: response_body.get_expired_at_as_u64(),
                offset: body_offset,
                size: content_length,
                base64ed_sha1,
                response_body,
            };
            initialized
                .resumed_records
                .persist(&record, &uploader.bucket_name()?, initialized.up_endpoints())
                .ok();
            Ok(MultiPartsV1UploaderUploadedPart::from_record(record, false))
        }
    }

    fn complete_parts(&self, initialized: &Self::InitializedParts, parts: &[Self::UploadedPart]) -> ApiResult<Value> {
        let file_size = get_file_size_from_uploaded_parts(parts);
        let upload_token_signer = self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
        let params = make_mkfile_path_params_from_initialized_parts(&initialized.params, file_size);
        let mkfile = self.storage().resumable_upload_v1_make_file();
        let body = make_mkfile_request_body_from_uploaded_parts(parts.to_vec());
        return _complete_parts(
            self,
            mkfile.new_request(initialized.up_endpoints(), params, upload_token_signer.as_ref()),
            &initialized.source,
            body,
        );

        fn _complete_parts<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a, D: DataSource<H>>(
            uploader: &'a MultiPartsV1Uploader<H>,
            mut request: SyncMkFileRequestBuilder<'a, E>,
            source: &D,
            body: String,
        ) -> ApiResult<Value> {
            uploader.before_request_call(request.parts_mut())?;
            let content_length = body.len() as u64;
            let mut response_result = request.call(Cursor::new(body), content_length);
            uploader.after_response_call(&mut response_result)?;
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let body = response_result?.into_body();
            uploader.try_to_delete_records(&source).ok();
            Ok(body.into())
        }
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    type AsyncInitializedParts = MultiPartsV1UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>;

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    type AsyncUploadedPart = MultiPartsV1UploaderUploadedPart;

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_initialize_parts<D: AsyncDataSource<Self::HashAlgorithm> + 'static>(
        &self,
        source: D,
        params: ObjectParams,
    ) -> BoxFuture<ApiResult<Self::AsyncInitializedParts>> {
        Box::pin(async move {
            let up_endpoints = self.async_get_up_endpoints(&params).await?;
            let resumed_records = self.async_resume_or_create_records(&source, up_endpoints).await?;
            Ok(Self::AsyncInitializedParts {
                source: Box::new(source),
                params,
                resumed_records,
                progresses: Default::default(),
            })
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn try_to_async_resume_parts<D: AsyncDataSource<Self::HashAlgorithm> + 'static>(
        &self,
        source: D,
        params: ObjectParams,
    ) -> BoxFuture<Option<Self::AsyncInitializedParts>> {
        Box::pin(async move {
            if let Some(source_key) = source.source_key().await.ok().flatten() {
                self.try_to_async_resume_records(&source_key)
                    .await
                    .ok()
                    .flatten()
                    .map(|resumed_records| Self::AsyncInitializedParts {
                        source: Box::new(source),
                        params,
                        resumed_records,
                        progresses: Default::default(),
                    })
            } else {
                None
            }
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_reinitialize_parts<'r>(
        &'r self,
        initialized: &'r mut Self::AsyncInitializedParts,
        options: ReinitializeOptions,
    ) -> BoxFuture<'r, ApiResult<()>> {
        Box::pin(async move {
            initialized.source.reset().await?;
            let up_endpoints = options.async_get_up_endpoints(self, initialized).await?;
            initialized.resumed_records = self.async_create_new_records(&initialized.source, up_endpoints).await?;
            initialized.progresses = Default::default();
            Ok(())
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_upload_part<'r>(
        &'r self,
        initialized: &'r Self::AsyncInitializedParts,
        data_partitioner_provider: &'r dyn DataPartitionProvider,
    ) -> BoxFuture<'r, ApiResult<Option<Self::AsyncUploadedPart>>> {
        return Box::pin(async move {
            let data_partitioner_provider = normalize_data_partitioner_provider(data_partitioner_provider);
            let total_size = initialized.source.total_size().await?;
            if let Some(mut reader) = initialized.source.slice(data_partitioner_provider.part_size()).await? {
                if let Some(part_size) = NonZeroU64::new(reader.len().await?) {
                    let progresses_key = initialized.progresses.add_new_part(part_size.into());
                    if let Some(uploaded_part) = _could_resume(initialized, &mut reader, part_size).await {
                        self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                        Ok(Some(uploaded_part))
                    } else {
                        let params = MkBlkPathParams::default().set_block_size_as_u64(part_size.into());
                        let upload_token_signer =
                            self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
                        let mkblk = self.storage().resumable_upload_v1_make_block();
                        let uploaded_result = _upload_part(
                            self,
                            mkblk.new_async_request(initialized.up_endpoints(), params, upload_token_signer.as_ref()),
                            reader,
                            part_size,
                            &progresses_key,
                            initialized,
                            &data_partitioner_provider,
                        )
                        .await;
                        match uploaded_result {
                            Ok(uploaded_part) => {
                                self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                                Ok(Some(uploaded_part))
                            }
                            Err(err) => {
                                self.after_part_uploaded(&progresses_key, total_size, None).ok();
                                Err(err)
                            }
                        }
                    }
                } else {
                    Ok(None)
                }
            } else {
                Ok(None)
            }
        });

        async fn _could_resume<H: Digest>(
            initialized: &MultiPartsV1UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>,
            data_reader: &mut AsyncDataSourceReader,
            part_size: NonZeroU64,
        ) -> Option<MultiPartsV1UploaderUploadedPart> {
            let offset = data_reader.offset();
            OptionFuture::from(initialized.resumed_records.take(offset).map(|record| async move {
                if record.size == part_size
                    && record.expired_at() > SystemTime::now()
                    && Some(record.base64ed_sha1.as_str()) == sha1_of_async_reader(data_reader).await.ok().as_deref()
                {
                    Some(MultiPartsV1UploaderUploadedPart {
                        response_body: record.response_body.to_owned(),
                        uploaded_size: record.size,
                        resumed: true,
                        offset,
                    })
                } else {
                    None
                }
            }))
            .await
            .flatten()
        }

        #[allow(clippy::too_many_arguments)]
        async fn _upload_part<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV1Uploader<H>,
            mut request: AsyncMkBlkRequestBuilder<'a, E>,
            mut body: AsyncDataSourceReader,
            content_length: NonZeroU64,
            progresses_key: &'a ProgressesKey,
            initialized: &'a MultiPartsV1UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>,
            data_partitioner_provider: &'a dyn DataPartitionProvider,
        ) -> ApiResult<MultiPartsV1UploaderUploadedPart> {
            let total_size = initialized.source.total_size().await?;
            request.on_uploading_progress(move |_, transfer| {
                progresses_key.update_part(transfer.transferred_bytes());
                uploader.callbacks.upload_progress(&UploadingProgressInfo::new(
                    progresses_key.current_uploaded(),
                    total_size,
                ))
            });
            uploader.before_request_call(request.parts_mut())?;
            let body_offset = body.offset();
            let base64ed_sha1 = sha1_of_async_reader(&mut body).await?;
            let begin_at = Instant::now();
            let mut response_result = request.call(body, content_length.get()).await;
            let elapsed = begin_at.elapsed();
            uploader.after_response_call(&mut response_result)?;
            let mut feedback_builder =
                DataPartitionProviderFeedback::builder(content_length.into(), elapsed, initialized.params.extensions());
            if let Some(err) = response_result.as_ref().err() {
                feedback_builder.error(err);
            }
            data_partitioner_provider.feedback(feedback_builder.build());
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let response_body = response_result?.into_body();
            let record = MultiPartsV1ResumableRecorderRecord {
                expired_timestamp: response_body.get_expired_at_as_u64(),
                offset: body_offset,
                size: content_length,
                base64ed_sha1,
                response_body,
            };
            initialized
                .resumed_records
                .async_persist(
                    &record,
                    &uploader.async_bucket_name().await?,
                    initialized.up_endpoints(),
                )
                .await
                .ok();
            Ok(MultiPartsV1UploaderUploadedPart::from_record(record, false))
        }

More examples

Hide additional examples

src/multi_parts_uploader/v2.rs (line 423)

        fn _upload_part<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: SyncUploadPartRequestBuilder<'a, E>,
            mut body: DataSourceReader,
            content_length: NonZeroU64,
            initialized: &'a MultiPartsV2UploaderInitializedObject<Box<dyn DataSource<H>>>,
            progresses_key: &'a ProgressesKey,
            data_partitioner_provider: &'a dyn DataPartitionProvider,
        ) -> ApiResult<MultiPartsV2UploaderUploadedPart> {
            let total_size = initialized.source.total_size()?;
            request.on_uploading_progress(move |_, transfer| {
                progresses_key.update_part(transfer.transferred_bytes());
                uploader.callbacks.upload_progress(&UploadingProgressInfo::new(
                    progresses_key.current_uploaded(),
                    total_size,
                ))
            });
            uploader.before_request_call(request.parts_mut())?;
            let body_offset = body.offset();
            let part_number = body.part_number();
            let base64ed_sha1 = sha1_of_sync_reader(&mut body)?;
            let begin_at = Instant::now();
            let mut response_result = request.call(body, content_length.get());
            let elapsed = begin_at.elapsed();
            uploader.after_response_call(&mut response_result)?;
            let mut feedback_builder =
                DataPartitionProviderFeedback::builder(content_length.into(), elapsed, initialized.params.extensions());
            if let Some(err) = response_result.as_ref().err() {
                feedback_builder.error(err);
            }
            data_partitioner_provider.feedback(feedback_builder.build());
            let response_body = response_result?.into_body();
            let record = MultiPartsV2ResumableRecorderRecord {
                response_body,
                offset: body_offset,
                size: content_length,
                base64ed_sha1,
                part_number,
            };
            initialized
                .resumed_records
                .persist(
                    &initialized.info,
                    &record,
                    &uploader.bucket_name()?,
                    initialized.params.object_name(),
                    initialized.up_endpoints(),
                )
                .ok();
            Ok(MultiPartsV2UploaderUploadedPart::from_record(record, false))
        }
    }

    fn complete_parts(&self, initialized: &Self::InitializedParts, parts: &[Self::UploadedPart]) -> ApiResult<Value> {
        let upload_token_signer = self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
        let path_params = make_complete_parts_path_params_from_initialized_params(
            self.bucket_name()?.to_string(),
            &initialized.params,
            initialized.info.upload_id.to_owned(),
        );
        let body = make_complete_parts_request_body_from_initialized_params(&initialized.params, parts.to_vec());
        let complete_parts = self.storage().resumable_upload_v2_complete_multipart_upload();
        return _complete_parts(
            self,
            complete_parts.new_request(&initialized.up_endpoints(), path_params, upload_token_signer.as_ref()),
            &initialized.source,
            body,
        );

        fn _complete_parts<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a, D: DataSource<H>>(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: SyncCompletePartsRequestBuilder<'a, E>,
            source: &D,
            body: CompletePartsRequestBody,
        ) -> ApiResult<Value> {
            uploader.before_request_call(request.parts_mut())?;
            let mut response_result = request.call(&body);
            uploader.after_response_call(&mut response_result)?;
            if let Err(err) = &mut response_result {
                may_set_extensions_in_err(err);
            }
            let body = response_result?.into_body();
            uploader.try_to_delete_records(&source).ok();
            Ok(body.into())
        }
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    type AsyncInitializedParts = MultiPartsV2UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>;

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    type AsyncUploadedPart = MultiPartsV2UploaderUploadedPart;

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_initialize_parts<D: AsyncDataSource<Self::HashAlgorithm> + 'static>(
        &self,
        source: D,
        params: ObjectParams,
    ) -> BoxFuture<ApiResult<Self::AsyncInitializedParts>> {
        Box::pin(async move {
            let (info, resumed_records) = self
                .async_resume_or_create_records(&source, &params, self.async_get_up_endpoints(&params).await?)
                .await?;
            let info = if let Some(info) = info {
                info
            } else {
                self.async_call_initialize_parts(&params, &resumed_records.up_endpoints)
                    .await?
            };

            Ok(Self::AsyncInitializedParts {
                source: Box::new(source),
                params,
                info,
                resumed_records,
                progresses: Default::default(),
            })
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn try_to_async_resume_parts<D: AsyncDataSource<Self::HashAlgorithm> + 'static>(
        &self,
        source: D,
        params: ObjectParams,
    ) -> BoxFuture<Option<Self::AsyncInitializedParts>> {
        Box::pin(async move {
            if let Some(source_key) = source.source_key().await.ok().flatten() {
                self.try_to_async_resume_records(&source_key, &params)
                    .await
                    .ok()
                    .flatten()
                    .map(|(info, resumed_records)| Self::AsyncInitializedParts {
                        source: Box::new(source),
                        params,
                        info,
                        resumed_records,
                        progresses: Default::default(),
                    })
            } else {
                None
            }
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_reinitialize_parts<'r>(
        &'r self,
        initialized: &'r mut Self::AsyncInitializedParts,
        options: ReinitializeOptions,
    ) -> BoxFuture<ApiResult<()>> {
        Box::pin(async move {
            initialized.source.reset().await?;
            let up_endpoints = options.async_get_up_endpoints(self, initialized).await?;
            initialized.resumed_records = self.async_create_new_records(&initialized.source, up_endpoints).await?;
            initialized.info = self
                .async_call_initialize_parts(&initialized.params, initialized.up_endpoints())
                .await?;
            initialized.progresses = Default::default();
            Ok(())
        })
    }

    #[cfg(feature = "async")]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    fn async_upload_part<'r>(
        &'r self,
        initialized: &'r Self::AsyncInitializedParts,
        data_partitioner_provider: &'r dyn DataPartitionProvider,
    ) -> BoxFuture<'r, ApiResult<Option<Self::AsyncUploadedPart>>> {
        return Box::pin(async move {
            let data_partitioner_provider = normalize_data_partitioner_provider(data_partitioner_provider);
            let total_size = initialized.source.total_size().await?;
            if let Some(mut reader) = initialized.source.slice(data_partitioner_provider.part_size()).await? {
                if let Some(part_size) = NonZeroU64::new(reader.len().await?) {
                    let progresses_key = initialized.progresses.add_new_part(part_size.get());
                    if let Some(uploaded_part) = _could_resume(initialized, &mut reader, part_size).await {
                        self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                        Ok(Some(uploaded_part))
                    } else {
                        let path_params = make_upload_part_path_params_from_initialized_params(
                            self.bucket_name()?.to_string(),
                            &initialized.params,
                            initialized.info.upload_id.to_owned(),
                            reader.part_number(),
                        );
                        let upload_token_signer =
                            self.make_upload_token_signer(initialized.params.object_name().map(|n| n.into()));
                        let upload_part = self.storage().resumable_upload_v2_upload_part();
                        let uploaded_result = _upload_part(
                            self,
                            upload_part.new_async_request(
                                initialized.up_endpoints(),
                                path_params,
                                upload_token_signer.as_ref(),
                            ),
                            reader,
                            part_size,
                            initialized,
                            &progresses_key,
                            &data_partitioner_provider,
                        )
                        .await;
                        match uploaded_result {
                            Ok(uploaded_part) => {
                                self.after_part_uploaded(&progresses_key, total_size, Some(&uploaded_part))?;
                                Ok(Some(uploaded_part))
                            }
                            Err(err) => {
                                self.after_part_uploaded(&progresses_key, total_size, None).ok();
                                Err(err)
                            }
                        }
                    }
                } else {
                    Ok(None)
                }
            } else {
                Ok(None)
            }
        });

        async fn _could_resume<H: Digest>(
            initialized: &MultiPartsV2UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>,
            data_reader: &mut AsyncDataSourceReader,
            part_size: NonZeroU64,
        ) -> Option<MultiPartsV2UploaderUploadedPart> {
            let offset = data_reader.offset();
            OptionFuture::from(initialized.resumed_records.take(offset).map(|record| async move {
                if record.size == part_size
                    && record.part_number == data_reader.part_number()
                    && Some(record.base64ed_sha1.as_str()) == sha1_of_async_reader(data_reader).await.ok().as_deref()
                {
                    Some(MultiPartsV2UploaderUploadedPart {
                        response_body: record.response_body.to_owned(),
                        uploaded_size: record.size,
                        part_number: record.part_number,
                        resumed: true,
                        offset,
                    })
                } else {
                    None
                }
            }))
            .await
            .flatten()
        }

        async fn _upload_part<'a, H: Digest + Send + 'static, E: EndpointsProvider + Clone + 'a>(
            uploader: &'a MultiPartsV2Uploader<H>,
            mut request: AsyncUploadPartRequestBuilder<'a, E>,
            mut body: AsyncDataSourceReader,
            content_length: NonZeroU64,
            initialized: &'a MultiPartsV2UploaderInitializedObject<Box<dyn AsyncDataSource<H>>>,
            progresses_key: &'a ProgressesKey,
            data_partitioner_provider: &'a dyn DataPartitionProvider,
        ) -> ApiResult<MultiPartsV2UploaderUploadedPart> {
            let total_size = initialized.source.total_size().await?;
            request.on_uploading_progress(move |_, transfer| {
                progresses_key.update_part(transfer.transferred_bytes());
                uploader.callbacks.upload_progress(&UploadingProgressInfo::new(
                    progresses_key.current_uploaded(),
                    total_size,
                ))
            });
            uploader.before_request_call(request.parts_mut())?;
            let body_offset = body.offset();
            let part_number = body.part_number();
            let base64ed_sha1 = sha1_of_async_reader(&mut body).await?;
            let begin_at = Instant::now();
            let mut response_result = request.call(body, content_length.get()).await;
            let elapsed = begin_at.elapsed();
            uploader.after_response_call(&mut response_result)?;
            let mut feedback_builder =
                DataPartitionProviderFeedback::builder(content_length.into(), elapsed, initialized.params.extensions());
            if let Some(err) = response_result.as_ref().err() {
                feedback_builder.error(err);
            }
            data_partitioner_provider.feedback(feedback_builder.build());
            let response_body = response_result?.into_body();
            let record = MultiPartsV2ResumableRecorderRecord {
                response_body,
                offset: body_offset,
                size: content_length,
                base64ed_sha1,
                part_number,
            };
            initialized
                .resumed_records
                .async_persist(
                    &initialized.info,
                    &record,
                    &uploader.async_bucket_name().await?,
                    initialized.params.object_name(),
                    initialized.up_endpoints(),
                )
                .await
                .ok();
            Ok(MultiPartsV2UploaderUploadedPart::from_record(record, false))
        }

Trait Implementations§

impl Clone for ObjectParams

fn clone(&self) -> ObjectParams

Returns a copy of the value. Read more

1.0.0 · source§

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source. Read more

impl Debug for ObjectParams

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter. Read more

impl Default for ObjectParams

fn default() -> ObjectParams

Returns the “default value” for a type. Read more

impl From<AutoUploaderObjectParams> for ObjectParams

fn from(auto_uploader_object_params: AutoUploaderObjectParams) -> Self

Converts to this type from the input type.

impl From<ObjectParams> for AutoUploaderObjectParams

fn from(object_params: ObjectParams) -> Self

Converts to this type from the input type.

Auto Trait Implementations§

impl !UnwindSafe for ObjectParams

Blanket Implementations§

impl<T> Any for Twhere
T: 'static + ?Sized,

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more

impl<T> Borrow<T> for Twhere
T: ?Sized,

const: unstable · source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more

impl<T> BorrowMut<T> for Twhere
T: ?Sized,

const: unstable · source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more

impl<T> Conv for T

fn conv<T>(self) -> Twhere
Self: Into<T>,

Converts self into T using Into<T>. Read more

impl<T> DynClone for Twhere
T: Clone,

fn __clone_box(&self, _: Private) -> *mut ()

impl<T> From<T> for T

const: unstable · source§

fn from(t: T) -> T

Returns the argument unchanged.

impl<T> Instrument for T

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more

impl<T> Instrument for T

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more

impl<T, U> Into for Twhere
U: From<T>,

const: unstable · source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

impl<T> Pipe for Twhere
T: ?Sized,

fn pipe<R>(self, func: impl FnOnce(Self) -> R) -> Rwhere
Self: Sized,

Pipes by value. This is generally the method you want to use. Read more

fn pipe_ref<'a, R>(&'a self, func: impl FnOnce(&'a Self) -> R) -> Rwhere
R: 'a,

Borrows self and passes that borrow into the pipe function. Read more

fn pipe_ref_mut<'a, R>(&'a mut self, func: impl FnOnce(&'a mut Self) -> R) -> Rwhere
R: 'a,

Mutably borrows self and passes that borrow into the pipe function. Read more

fn pipe_borrow<'a, B, R>(&'a self, func: impl FnOnce(&'a B) -> R) -> Rwhere
 Self: Borrow,
 B: 'a + ?Sized,
 R: 'a,

Borrows self, then passes self.borrow() into the pipe function. Read more

fn pipe_borrow_mut<'a, B, R>(
 &'a mut self,
 func: impl FnOnce(&'a mut B) -> R
) -> Rwhere
 Self: BorrowMut,
 B: 'a + ?Sized,
 R: 'a,

Mutably borrows self, then passes self.borrow_mut() into the pipe function. Read more

fn pipe_as_ref<'a, U, R>(&'a self, func: impl FnOnce(&'a U) -> R) -> Rwhere
 Self: AsRef,
 U: 'a + ?Sized,
 R: 'a,

Borrows self, then passes self.as_ref() into the pipe function.

fn pipe_as_mut<'a, U, R>(&'a mut self, func: impl FnOnce(&'a mut U) -> R) -> Rwhere
 Self: AsMut,
 U: 'a + ?Sized,
 R: 'a,

Mutably borrows self, then passes self.as_mut() into the pipe function.

fn pipe_deref<'a, T, R>(&'a self, func: impl FnOnce(&'a T) -> R) -> Rwhere
 Self: Deref<Target = T>,
 T: 'a + ?Sized,
 R: 'a,

Borrows self, then passes self.deref() into the pipe function.

fn pipe_deref_mut<'a, T, R>(&'a mut self, func: impl FnOnce(&'a mut T) -> R) -> Rwhere
 Self: DerefMut<Target = T> + Deref,
 T: 'a + ?Sized,
 R: 'a,

Mutably borrows self, then passes self.deref_mut() into the pipe function.

impl<T> Pointable for T

const ALIGN: usize = mem::align_of::<T>()

The alignment of pointer.

type Init = T

The type for initializers.

unsafe fn init(init: <T as Pointable>::Init) -> usize

Initializes a with the given initializer. Read more

unsafe fn deref<'a>(ptr: usize) -> &'a T

Dereferences the given pointer. Read more

unsafe fn deref_mut<'a>(ptr: usize) -> &'a mut T

Mutably dereferences the given pointer. Read more

unsafe fn drop(ptr: usize)

Drops the object pointed to by the given pointer. Read more

impl<T> Same<T> for T

type Output = T

Should always be Self

impl<T> Tap for T

fn tap(self, func: impl FnOnce(&Self)) -> Self

Immutable access to a value. Read more

fn tap_mut(self, func: impl FnOnce(&mut Self)) -> Self

Mutable access to a value. Read more

fn tap_borrow(self, func: impl FnOnce(&B)) -> Selfwhere
Self: Borrow,
B: ?Sized,

Immutable access to the Borrow of a value. Read more

fn tap_borrow_mut(self, func: impl FnOnce(&mut B)) -> Selfwhere
Self: BorrowMut,
B: ?Sized,

Mutable access to the BorrowMut of a value. Read more

fn tap_ref<R>(self, func: impl FnOnce(&R)) -> Selfwhere
Self: AsRef<R>,
R: ?Sized,

Immutable access to the AsRef<R> view of a value. Read more

fn tap_ref_mut<R>(self, func: impl FnOnce(&mut R)) -> Selfwhere
Self: AsMut<R>,
R: ?Sized,

Mutable access to the AsMut<R> view of a value. Read more

fn tap_deref<T>(self, func: impl FnOnce(&T)) -> Selfwhere
Self: Deref<Target = T>,
T: ?Sized,

Immutable access to the Deref::Target of a value. Read more

fn tap_deref_mut<T>(self, func: impl FnOnce(&mut T)) -> Selfwhere
Self: DerefMut<Target = T> + Deref,
T: ?Sized,

Mutable access to the Deref::Target of a value. Read more

fn tap_dbg(self, func: impl FnOnce(&Self)) -> Self

Calls .tap() only in debug builds, and is erased in release builds.

fn tap_mut_dbg(self, func: impl FnOnce(&mut Self)) -> Self

Calls .tap_mut() only in debug builds, and is erased in release builds.

fn tap_borrow_dbg(self, func: impl FnOnce(&B)) -> Selfwhere
Self: Borrow,
B: ?Sized,

Calls .tap_borrow() only in debug builds, and is erased in release builds.

fn tap_borrow_mut_dbg(self, func: impl FnOnce(&mut B)) -> Selfwhere
Self: BorrowMut,
B: ?Sized,

Calls .tap_borrow_mut() only in debug builds, and is erased in release builds.

fn tap_ref_dbg<R>(self, func: impl FnOnce(&R)) -> Selfwhere
Self: AsRef<R>,
R: ?Sized,

Calls .tap_ref() only in debug builds, and is erased in release builds.

fn tap_ref_mut_dbg<R>(self, func: impl FnOnce(&mut R)) -> Selfwhere
Self: AsMut<R>,
R: ?Sized,

Calls .tap_ref_mut() only in debug builds, and is erased in release builds.

fn tap_deref_dbg<T>(self, func: impl FnOnce(&T)) -> Selfwhere
Self: Deref<Target = T>,
T: ?Sized,

Calls .tap_deref() only in debug builds, and is erased in release builds.

fn tap_deref_mut_dbg<T>(self, func: impl FnOnce(&mut T)) -> Selfwhere
Self: DerefMut<Target = T> + Deref,
T: ?Sized,

Calls .tap_deref_mut() only in debug builds, and is erased in release builds.

impl<T> ToOwned for Twhere
T: Clone,

type Owned = T

The resulting type after obtaining ownership.

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more

impl<T> TryConv for T

fn try_conv<T>(self) -> Result<T, Self::Error>where
Self: TryInto<T>,

Attempts to convert self into T using TryInto<T>. Read more

impl<T, U> TryFrom for Twhere
U: Into<T>,

type Error = Infallible

The type returned in the event of a conversion error.

const: unstable · source§

fn try_from(value: U) -> Result<T, <T as TryFrom>::Error>

Performs the conversion.

impl<T, U> TryInto for Twhere
U: TryFrom<T>,

type Error = >::Error

The type returned in the event of a conversion error.

const: unstable · source§

fn try_into(self) -> Result<U, >::Error>

Performs the conversion.

impl<V, T> VZip<V> for Twhere
V: MultiLane<T>,

fn vzip(self) -> V

impl<T> WithSubscriber for T

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>where
S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more

impl<T> WithSubscriber for T

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>where
S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more