Struct qiniu_upload_manager::DataPartitionProviderFeedback

pub struct DataPartitionProviderFeedback<'f> { /* private fields */ }

分片大小提供者反馈

反馈给提供者分片的效果，包含对象大小，花费时间，以及错误信息。

Implementations

impl<'f> DataPartitionProviderFeedback<'f>

pub fn builder(
    part_size: PartSize,
    elapsed: Duration,
    extensions: &'f Extensions
) -> DataPartitionProviderFeedbackBuilder<'f>

创建分片大小提供者反馈构建器

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

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

pub fn part_size(&self) -> PartSize

获取分片大小

pub fn elapsed(&self) -> Duration

获取花费时间

pub fn extensions(&self) -> &Extensions

获取扩展信息

pub fn error(&self) -> Option<&'f ResponseError>

获取错误信息

Trait Implementations

impl<'f> Clone for DataPartitionProviderFeedback<'f>

fn clone(&self) -> DataPartitionProviderFeedback<'f>

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<'f> Debug for DataPartitionProviderFeedback<'f>

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

Formats the value using the given formatter.