Struct qiniu_upload_manager::UploadingProgressInfo

pub struct UploadingProgressInfo { /* private fields */ }

上传进度信息

Implementations

impl UploadingProgressInfo

pub fn new(transferred_bytes: u64, total_bytes: Option<u64>) -> Self

创建上传进度信息

Examples found in repository

src/callbacks.rs (line 186)

    fn from(t: &'a TransferProgressInfo<'a>) -> Self {
        Self::new(t.transferred_bytes(), Some(t.total_bytes()))
    }
}

impl From<TransferProgressInfo<'_>> for UploadingProgressInfo {
    #[inline]
    fn from(t: TransferProgressInfo<'_>) -> Self {
        Self::new(t.transferred_bytes(), Some(t.total_bytes()))
    }

src/multi_parts_uploader/v1.rs (lines 357-360)

        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
}

fn get_file_size_from_uploaded_parts(parts: &[MultiPartsV1UploaderUploadedPart]) -> u64 {
    parts
        .iter()
        .map(|uploaded_part| uploaded_part.uploaded_size.get())
        .sum()
}

fn make_mkfile_request_body_from_uploaded_parts(mut parts: Vec<MultiPartsV1UploaderUploadedPart>) -> String {
    parts.sort_by_key(|part| part.offset);
    parts
        .iter()
        .map(|part| part.response_body.get_ctx_as_str())
        .enumerate()
        .fold(String::new(), |mut joined, (i, ctx)| {
            if i > 0 {
                joined += ",";
            }
            joined += ctx;
            joined
        })
}

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() == 701 => {
            err.extensions_mut().insert(PartsExpiredError);
        }
        _ => {}
    }
}

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

impl<H: Digest + Send + 'static> MultiPartsV1Uploader<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<&MultiPartsV1UploaderUploadedPart>,
    ) -> 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)
    }

src/multi_parts_uploader/v2.rs (lines 409-412)

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

pub fn transferred_bytes(&self) -> u64

获取已传输的字节数

pub fn total_bytes(&self) -> Option<u64>

获取总字节数

Trait Implementations

impl Clone for UploadingProgressInfo

fn clone(&self) -> UploadingProgressInfo

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for UploadingProgressInfo

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

Formats the value using the given formatter.

impl<'a> From<&'a TransferProgressInfo<'a>> for UploadingProgressInfo

fn from(t: &'a TransferProgressInfo<'a>) -> Self

Converts to this type from the input type.

impl From<TransferProgressInfo<'_>> for UploadingProgressInfo

fn from(t: TransferProgressInfo<'_>) -> Self

Converts to this type from the input type.

impl Copy for UploadingProgressInfo