Struct qiniu_objects_manager::ObjectsManager

impl ObjectsManager

pub fn builder(
credential: impl CredentialProvider + 'static
) -> ObjectsManagerBuilder

创建七牛对象管理构建器

必须传入认证信息提供者

Examples found in repository ?

src/objects_manager.rs (line 37)

36
37
38

    pub fn new(credential: impl CredentialProvider + 'static) -> Self {
        Self::builder(credential).build()
    }

pub fn new(credential: impl CredentialProvider + 'static) -> Self

创建七牛对象管理器

必须传入认证信息提供者

pub fn client(&self) -> &QiniuApiClient

获取七牛 API 调用客户端

Examples found in repository ?

src/batch_operations.rs (line 209)

    fn make_request(&self) -> ApiResult<BatchOpsSyncRequestBuilder<'a, RefRegionProviderEndpoints<'a>>> {
        let request = self
            .operations
            .bucket
            .objects_manager()
            .client()
            .storage()
            .batch_ops()
            .new_request(
                RegionsProviderEndpoints::new(self.operations.bucket.region_provider()?),
                self.operations.bucket.objects_manager().credential(),
            );
        Ok(request)
    }

    fn call_request(
        &mut self,
        mut request: BatchOpsSyncRequestBuilder<'_, RefRegionProviderEndpoints>,
        request_body: RequestBody,
    ) -> ApiResult<Response<ResponseBody>> {
        self.operations
            .callbacks
            .before_request(request.parts_mut())
            .map_err(make_callback_error)?;
        let mut response_result = request.call(request_body);
        self.operations
            .callbacks
            .after_response(&mut response_result)
            .map_err(make_callback_error)?;
        response_result
    }

    fn handle_response(&mut self, response_body: ResponseBody) -> Option<ApiResult<OperationResponseData>> {
        let responses = response_body.to_operation_response_vec();
        self.buffer
            .extend(responses.into_iter().map(from_response_to_response_data_result));
        self.buffer.pop_front()
    }

    fn generate_request_body(&mut self) -> Option<RequestBody> {
        let mut request_body = RequestBody::default();
        let mut operation_count = 0usize;
        for _ in 0..self
            .operations
            .batch_size
            .as_ref()
            .map(|provider| provider.batch_size())
            .unwrap_or(DEFAULT_BATCH_SIZE)
        {
            if let Some(mut operation) = self.operations.operations.as_mut().and_then(|op| op.next()) {
                request_body = request_body.append_operations_as_str(operation.to_operation());
                operation_count += 1;
            } else {
                self.closed = true;
                break;
            }
        }
        if operation_count > 0 {
            Some(request_body)
        } else {
            None
        }
    }

    #[allow(dead_code)]
    fn assert() {
        assert_impl!(Send: Self);
        assert_impl!(Sync: Self);
    }
}

#[cfg(feature = "async")]
mod async_stream {
    use super::*;
    use futures::{future::BoxFuture, ready, FutureExt, Stream};
    use qiniu_apis::storage::batch_ops::AsyncRequestBuilder as BatchOpsAsyncRequestBuilder;
    use smart_default::SmartDefault;
    use std::{
        fmt::{self, Debug},
        io::Result as IOResult,
        pin::Pin,
        task::{Context, Poll},
    };

    /// 批量操作流
    ///
    /// 实现 [`futures::stream::Stream`] 接口，
    /// 在迭代过程中异步发起批量操作 API
    #[must_use]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    #[derive(Debug)]
    pub struct BatchOperationsStream<'a> {
        operations: BatchOperations<'a>,
        current_step: BatchOperationsStep<'a>,
        closed: bool,
    }

    #[derive(SmartDefault)]
    enum BatchOperationsStep<'a> {
        #[default]
        FromBuffer {
            buffer: VecDeque<ApiResult<OperationResponseData>>,
        },
        WaitForResponse {
            task: BoxFuture<'a, ApiResult<Response<ResponseBody>>>,
        },
        WaitForRegionProvider {
            task: BoxFuture<'a, IOResult<&'a dyn RegionsProvider>>,
        },
        Done,
    }

    impl Debug for BatchOperationsStep<'_> {
        #[inline]
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            match self {
                Self::FromBuffer { buffer } => f.debug_tuple("FromBuffer").field(buffer).finish(),
                Self::WaitForResponse { .. } => f.debug_tuple("WaitForResponse").finish(),
                Self::WaitForRegionProvider { .. } => f.debug_tuple("WaitForRegionProvider").finish(),
                Self::Done => f.debug_tuple("Done").finish(),
            }
        }
    }

    impl Stream for BatchOperationsStream<'_> {
        type Item = ApiResult<OperationResponseData>;

        #[inline]
        fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            match self.current_step {
                BatchOperationsStep::FromBuffer { .. } => self.read_from_buffer(cx),
                BatchOperationsStep::WaitForResponse { .. } => self.wait_for_response(cx),
                BatchOperationsStep::WaitForRegionProvider { .. } => self.wait_for_region(cx),
                BatchOperationsStep::Done { .. } => Poll::Ready(None),
            }
        }
    }

    impl<'a> BatchOperationsStream<'a> {
        pub(super) fn new(operations: BatchOperations<'a>) -> Self {
            Self {
                operations,
                current_step: Default::default(),
                closed: false,
            }
        }

        fn read_from_buffer(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let BatchOperationsStep::FromBuffer { buffer } = &mut self.current_step {
                if let Some(response) = buffer.pop_front() {
                    Poll::Ready(Some(response))
                } else if self.closed {
                    self.current_step = BatchOperationsStep::Done;
                    self.poll_next(cx)
                } else {
                    let bucket = self.operations.bucket;
                    self.current_step = BatchOperationsStep::WaitForRegionProvider {
                        task: Box::pin(async move { bucket.async_region_provider().await }),
                    };
                    self.poll_next(cx)
                }
            } else {
                unreachable!()
            }
        }

        fn wait_for_response(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let BatchOperationsStep::WaitForResponse { task } = &mut self.current_step {
                let mut response_result = ready!(task.poll_unpin(cx));
                if let Err(err) = self.operations.callbacks.after_response(&mut response_result) {
                    self.current_step = BatchOperationsStep::Done;
                    return Poll::Ready(Some(Err(make_callback_error(err))));
                }
                match response_result {
                    Ok(response) => {
                        self.current_step = BatchOperationsStep::FromBuffer {
                            buffer: response
                                .into_body()
                                .to_operation_response_vec()
                                .into_iter()
                                .map(from_response_to_response_data_result)
                                .collect(),
                        };
                        self.poll_next(cx)
                    }
                    Err(err) => {
                        self.current_step = BatchOperationsStep::Done;
                        Poll::Ready(Some(Err(err)))
                    }
                }
            } else {
                unreachable!()
            }
        }

        fn wait_for_region(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let BatchOperationsStep::WaitForRegionProvider { task } = &mut self.current_step {
                match ready!(task.poll_unpin(cx)) {
                    Ok(region_provider) => {
                        if let Some(request_body) = self.generate_request_body() {
                            let mut request = self.make_request(region_provider);
                            if let Err(err) = self.operations.callbacks.before_request(request.parts_mut()) {
                                self.current_step = BatchOperationsStep::Done;
                                Poll::Ready(Some(Err(make_callback_error(err))))
                            } else {
                                self.current_step = BatchOperationsStep::WaitForResponse {
                                    task: Box::pin(async move { request.call(request_body).await }),
                                };
                                self.poll_next(cx)
                            }
                        } else {
                            self.current_step = BatchOperationsStep::Done;
                            self.poll_next(cx)
                        }
                    }
                    Err(err) => {
                        self.current_step = BatchOperationsStep::Done;
                        Poll::Ready(Some(Err(err.into())))
                    }
                }
            } else {
                unreachable!()
            }
        }

        fn generate_request_body(&mut self) -> Option<RequestBody> {
            let mut request_body = RequestBody::default();
            let mut operation_count = 0usize;
            for _ in 0..self
                .operations
                .batch_size
                .as_ref()
                .map(|provider| provider.batch_size())
                .unwrap_or(DEFAULT_BATCH_SIZE)
            {
                if let Some(mut operation) = self.operations.operations.as_mut().and_then(|op| op.next()) {
                    request_body = request_body.append_operations_as_str(operation.to_operation());
                    operation_count += 1;
                } else {
                    self.closed = true;
                    break;
                }
            }
            if operation_count > 0 {
                Some(request_body)
            } else {
                None
            }
        }

        fn make_request(
            &self,
            region_provider: &'a dyn RegionsProvider,
        ) -> BatchOpsAsyncRequestBuilder<'a, RefRegionProviderEndpoints<'a>> {
            self.operations
                .bucket
                .objects_manager()
                .client()
                .storage()
                .batch_ops()
                .new_async_request(
                    RegionsProviderEndpoints::new(region_provider),
                    self.operations.bucket.objects_manager().credential(),
                )
        }

More examples

Hide additional examples

src/list.rs (line 310)

        fn v1_make_request<'a>(
            params: &mut ListParams<'a>,
        ) -> ApiResult<GetObjectsV1SyncRequestBuilder<'a, RefRegionProviderEndpoints<'a>>> {
            let mut request = params
                .bucket
                .objects_manager()
                .client()
                .storage()
                .get_objects()
                .new_request(
                    RegionsProviderEndpoints::new(params.bucket.region_provider()?),
                    params.bucket.objects_manager().credential(),
                );
            request.query_pairs(params.to_query_params());
            Ok(request)
        }

        fn v1_call_request(
            mut request: GetObjectsV1SyncRequestBuilder<'_, RefRegionProviderEndpoints>,
            callbacks: &mut Callbacks<'_>,
        ) -> ApiResult<Response<GetObjectsV1ResponseBody>> {
            callbacks
                .before_request(request.parts_mut())
                .map_err(make_callback_error)?;
            let mut response_result = request.call();
            callbacks
                .after_response(&mut response_result)
                .map_err(make_callback_error)?;
            response_result
        }

        fn v1_handle_response(
            body: GetObjectsV1ResponseBody,
            params: &mut ListParams<'_>,
            buffer: &mut VecDeque<ListedObjectEntry>,
        ) {
            params.marker.set(body.get_marker_as_str());
            let listed_object_entries = body.get_items().to_listed_object_entry_vec();
            params.limit.saturating_decrease(listed_object_entries.len());
            *buffer = listed_object_entries.into();
        }

        fn v2_next(
            params: &mut ListParams<'_>,
            callbacks: &mut Callbacks<'_>,
            step: &mut SyncListV2Step,
        ) -> Option<ApiResult<ListedObjectEntry>> {
            match step {
                SyncListV2Step::Start => match v2_call(params, callbacks) {
                    Ok(Some(mut lines)) => v2_read_entry_from_lines(params, &mut lines).tap_some(|result| {
                        if result.is_ok() {
                            *step = SyncListV2Step::Lines { lines };
                        } else {
                            *step = SyncListV2Step::Done;
                        }
                    }),
                    Ok(None) => {
                        *step = SyncListV2Step::Done;
                        None
                    }
                    Err(err) => {
                        *step = SyncListV2Step::Done;
                        Some(Err(err))
                    }
                },
                SyncListV2Step::Lines { lines } => match v2_read_entry_from_lines(params, lines) {
                    Some(Ok(entry)) => Some(Ok(entry)),
                    Some(Err(err)) => {
                        warn!("Read Error from ListV2 Response Body: {}", err);
                        *step = SyncListV2Step::Start;
                        v2_next(params, callbacks, step)
                    }
                    None => {
                        *step = SyncListV2Step::Start;
                        v2_next(params, callbacks, step)
                    }
                },
                SyncListV2Step::Done => None,
            }
        }

        fn v2_read_entry_from_lines(
            params: &mut ListParams<'_>,
            lines: &mut Lines<BufReader<SyncResponseBody>>,
        ) -> Option<ApiResult<ListedObjectEntry>> {
            if params.limit.exhausted() {
                return None;
            }
            loop {
                match lines.next() {
                    Some(Ok(line)) if line.is_empty() => {
                        continue;
                    }
                    Some(Ok(line)) => match serde_json::from_str::<ListedObjectEntryV2>(&line) {
                        Ok(parsed) => {
                            params.marker.set(parsed.marker.as_deref());
                            if let Some(item) = parsed.item {
                                params.limit.saturating_decrease(1);
                                return Some(Ok(item));
                            } else {
                                continue;
                            }
                        }
                        Err(err) => {
                            return Some(Err(err.into()));
                        }
                    },
                    Some(Err(err)) => {
                        return Some(Err(err.into()));
                    }
                    None => {
                        return None;
                    }
                }
            }
        }

        fn v2_call(
            params: &mut ListParams<'_>,
            callbacks: &mut Callbacks<'_>,
        ) -> ApiResult<Option<Lines<BufReader<SyncResponseBody>>>> {
            if params.have_done() {
                return Ok(None);
            }
            let request = v2_make_request(params)?;
            let response_result = v2_call_request(request, callbacks);
            Ok(Some(BufReader::new(response_result?.into_body()).lines()))
        }

        fn v2_make_request<'a>(
            params: &mut ListParams<'a>,
        ) -> ApiResult<GetObjectsV2SyncRequestBuilder<'a, RefRegionProviderEndpoints<'a>>> {
            let mut request = params
                .bucket
                .objects_manager()
                .client()
                .storage()
                .get_objects_v2()
                .new_request(
                    RegionsProviderEndpoints::new(params.bucket.region_provider()?),
                    params.bucket.objects_manager().credential(),
                );
            request.query_pairs(params.to_query_params());
            Ok(request)
        }

        fn v2_call_request(
            mut request: GetObjectsV2SyncRequestBuilder<'_, RefRegionProviderEndpoints>,
            callbacks: &mut Callbacks<'_>,
        ) -> ApiResult<Response<SyncResponseBody>> {
            callbacks
                .before_request(request.parts_mut())
                .map_err(make_callback_error)?;
            let mut response_result = request.call();
            callbacks
                .after_response(&mut response_result)
                .map_err(make_callback_error)?;
            response_result
        }
    }
}

#[cfg(feature = "async")]
mod async_list_stream {
    use super::*;
    use futures::{future::BoxFuture, io::Lines as AsyncLines, ready, AsyncBufReadExt, FutureExt, Stream, StreamExt};
    use std::{
        fmt::{self, Debug},
        io::Result as IOResult,
        pin::Pin,
        task::{Context, Poll},
    };

    enum AsyncListVersionWithStep<'a> {
        V1(AsyncListV1Step<'a>),
        V2(AsyncListV2Step<'a>),
    }

    impl From<ListVersion> for AsyncListVersionWithStep<'_> {
        fn from(version: ListVersion) -> Self {
            match version {
                ListVersion::V1 => Self::V1(Default::default()),
                ListVersion::V2 => Self::V2(Default::default()),
            }
        }
    }

    #[derive(SmartDefault)]
    enum AsyncListV1Step<'a> {
        #[default]
        FromBuffer {
            buffer: VecDeque<ListedObjectEntry>,
        },
        WaitForResponse {
            task: BoxFuture<'a, ApiResult<Response<GetObjectsV1ResponseBody>>>,
        },
        WaitForRegionProvider {
            task: BoxFuture<'a, IOResult<&'a dyn RegionsProvider>>,
        },
        Done,
    }

    impl Debug for AsyncListV1Step<'_> {
        #[inline]
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            match self {
                Self::FromBuffer { buffer } => f.debug_tuple("FromBuffer").field(buffer).finish(),
                Self::WaitForResponse { .. } => f.debug_tuple("WaitForResponse").finish(),
                Self::WaitForRegionProvider { .. } => f.debug_tuple("WaitForRegionProvider").finish(),
                Self::Done => f.debug_tuple("Done").finish(),
            }
        }
    }

    trait StreamWithMarker: Stream {
        fn marker(&self) -> Option<&str>;
    }
    type BoxStreamWithMarker<'a, T> = Pin<Box<dyn StreamWithMarker<Item = T> + Send + 'a>>;
    type ListedObjectEntryResultStream<'a> = BoxStreamWithMarker<'a, ApiResult<ListedObjectEntry>>;

    /// 对象列举流
    ///
    /// 实现 [`futures::stream::Stream`] 接口，
    /// 在迭代过程中异步发起 API 列举对象信息
    ///
    /// 可以通过 [`crate::ListBuilder::stream`] 方法获取该迭代器。
    #[must_use]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    pub struct ListStream<'a>(ListedObjectEntryResultStream<'a>);

    impl<'a> ListStream<'a> {
        pub(in super::super) fn new(
            bucket: &'a Bucket,
            limit: Option<usize>,
            prefix: Option<Cow<'a, str>>,
            marker: Option<Cow<'a, str>>,
            need_parts: bool,
            version: ListVersion,
            callbacks: Callbacks<'a>,
        ) -> Self {
            Self(match version {
                ListVersion::V1 => v1_next(bucket, limit, prefix, marker, need_parts, callbacks),
                ListVersion::V2 => v2_next(bucket, limit, prefix, marker, need_parts, callbacks),
            })
        }

        #[allow(dead_code)]
        fn assert() {
            assert_impl!(Send: Self);
            // assert_impl!(Sync: Self);
        }

        /// 获取上一次列举返回的位置标记
        pub fn marker(&self) -> Option<&str> {
            self.0.marker()
        }
    }

    impl Stream for ListStream<'_> {
        type Item = ApiResult<ListedObjectEntry>;

        #[inline]
        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            self.0.poll_next_unpin(cx)
        }
    }

    impl Debug for ListStream<'_> {
        #[inline]
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.debug_struct("ListStream").finish()
        }
    }

    #[derive(Debug)]
    struct ListV1Stream<'a> {
        params: ListParams<'a>,
        callbacks: Callbacks<'a>,
        current_step: AsyncListV1Step<'a>,
    }

    fn v1_next<'a>(
        bucket: &'a Bucket,
        limit: Option<usize>,
        prefix: Option<Cow<'a, str>>,
        marker: Option<Cow<'a, str>>,
        need_parts: bool,
        callbacks: Callbacks<'a>,
    ) -> ListedObjectEntryResultStream<'a> {
        let params = ListParams {
            bucket,
            prefix,
            need_parts,
            limit: Limit::new(limit, ListVersion::V1),
            marker: Marker::new(marker),
        };
        Box::pin(ListV1Stream {
            params,
            callbacks,
            current_step: Default::default(),
        })
    }

    impl Stream for ListV1Stream<'_> {
        type Item = ApiResult<ListedObjectEntry>;

        #[inline]
        fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            match self.current_step {
                AsyncListV1Step::FromBuffer { .. } => self.read_from_buffer(cx),
                AsyncListV1Step::WaitForResponse { .. } => self.wait_for_response(cx),
                AsyncListV1Step::WaitForRegionProvider { .. } => self.wait_for_region(cx),
                AsyncListV1Step::Done => Poll::Ready(None),
            }
        }
    }

    impl StreamWithMarker for ListV1Stream<'_> {
        fn marker(&self) -> Option<&str> {
            self.params.marker.as_ref()
        }
    }

    impl ListV1Stream<'_> {
        fn read_from_buffer(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV1Step::FromBuffer { buffer } = &mut self.current_step {
                if let Some(object) = buffer.pop_front() {
                    Poll::Ready(Some(Ok(object)))
                } else {
                    if self.params.have_done() {
                        self.current_step = AsyncListV1Step::Done;
                    } else {
                        let bucket = self.params.bucket;
                        self.current_step = AsyncListV1Step::WaitForRegionProvider {
                            task: Box::pin(async move { bucket.async_region_provider().await }),
                        };
                    }
                    self.poll_next(cx)
                }
            } else {
                unreachable!()
            }
        }

        fn wait_for_region(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV1Step::WaitForRegionProvider { task } = &mut self.current_step {
                match ready!(task.poll_unpin(cx)) {
                    Ok(region_provider) => {
                        let credential = self.params.bucket.objects_manager().credential();
                        let mut request = self
                            .params
                            .bucket
                            .objects_manager()
                            .client()
                            .storage()
                            .get_objects()
                            .new_async_request(RegionsProviderEndpoints::new(region_provider), credential);
                        request.query_pairs(self.params.to_query_params());
                        if let Err(err) = self.callbacks.before_request(request.parts_mut()) {
                            self.current_step = AsyncListV1Step::Done;
                            Poll::Ready(Some(Err(make_callback_error(err))))
                        } else {
                            self.current_step = AsyncListV1Step::WaitForResponse {
                                task: Box::pin(async move { request.call().await }),
                            };
                            self.poll_next(cx)
                        }
                    }
                    Err(err) => {
                        self.current_step = AsyncListV1Step::Done;
                        Poll::Ready(Some(Err(err.into())))
                    }
                }
            } else {
                unreachable!()
            }
        }

        fn wait_for_response(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV1Step::WaitForResponse { task } = &mut self.current_step {
                let mut response_result = ready!(task.poll_unpin(cx));
                if let Err(err) = self.callbacks.after_response(&mut response_result) {
                    self.current_step = AsyncListV1Step::Done;
                    Poll::Ready(Some(Err(make_callback_error(err))))
                } else {
                    match response_result {
                        Ok(response) => {
                            let body = response.into_body();
                            let listed_object_entries = body.get_items().to_listed_object_entry_vec();
                            self.params.marker.set(body.get_marker_as_str());
                            self.params.limit.saturating_decrease(listed_object_entries.len());
                            self.current_step = AsyncListV1Step::FromBuffer {
                                buffer: listed_object_entries.into(),
                            };
                            self.poll_next(cx)
                        }
                        Err(err) => {
                            self.current_step = AsyncListV1Step::Done;
                            Poll::Ready(Some(Err(err)))
                        }
                    }
                }
            } else {
                unreachable!()
            }
        }
    }

    #[derive(SmartDefault)]
    enum AsyncListV2Step<'a> {
        #[default]
        Start,
        WaitForRegionProvider {
            task: BoxFuture<'a, IOResult<&'a dyn RegionsProvider>>,
        },
        WaitForResponse {
            task: BoxFuture<'a, ApiResult<Response<AsyncResponseBody>>>,
        },
        WaitForEntries {
            lines: AsyncLines<AsyncBufReader<AsyncResponseBody>>,
            empty: bool,
        },
        Done,
    }

    struct ListV2Stream<'a> {
        params: ListParams<'a>,
        callbacks: Callbacks<'a>,
        current_step: AsyncListV2Step<'a>,
    }

    #[allow(clippy::too_many_arguments)]
    fn v2_next<'a>(
        bucket: &'a Bucket,
        limit: Option<usize>,
        prefix: Option<Cow<'a, str>>,
        marker: Option<Cow<'a, str>>,
        need_parts: bool,
        callbacks: Callbacks<'a>,
    ) -> ListedObjectEntryResultStream<'a> {
        let params = ListParams {
            bucket,
            prefix,
            need_parts,
            limit: Limit::new(limit, ListVersion::V2),
            marker: Marker::new(marker),
        };
        Box::pin(ListV2Stream {
            params,
            callbacks,
            current_step: Default::default(),
        })
    }

    impl Stream for ListV2Stream<'_> {
        type Item = ApiResult<ListedObjectEntry>;

        #[inline]
        fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            match self.current_step {
                AsyncListV2Step::Start { .. } => self.start(cx),
                AsyncListV2Step::WaitForResponse { .. } => self.wait_for_response(cx),
                AsyncListV2Step::WaitForRegionProvider { .. } => self.wait_for_region(cx),
                AsyncListV2Step::WaitForEntries { .. } => self.wait_for_entries(cx),
                AsyncListV2Step::Done => Poll::Ready(None),
            }
        }
    }

    impl StreamWithMarker for ListV2Stream<'_> {
        fn marker(&self) -> Option<&str> {
            self.params.marker.as_ref()
        }
    }

    impl ListV2Stream<'_> {
        fn start(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV2Step::Start { .. } = &mut self.current_step {
                if self.params.have_done() {
                    self.current_step = AsyncListV2Step::Done;
                } else {
                    let bucket = self.params.bucket;
                    self.current_step = AsyncListV2Step::WaitForRegionProvider {
                        task: Box::pin(async move { bucket.async_region_provider().await }),
                    };
                }
                self.poll_next(cx)
            } else {
                unreachable!()
            }
        }

        fn wait_for_region(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV2Step::WaitForRegionProvider { task } = &mut self.current_step {
                match ready!(task.poll_unpin(cx)) {
                    Ok(region_provider) => {
                        let credential = self.params.bucket.objects_manager().credential();
                        let mut request = self
                            .params
                            .bucket
                            .objects_manager()
                            .client()
                            .storage()
                            .get_objects_v2()
                            .new_async_request(RegionsProviderEndpoints::new(region_provider), credential);
                        request.query_pairs(self.params.to_query_params());
                        if let Err(err) = self.callbacks.before_request(request.parts_mut()) {
                            self.current_step = AsyncListV2Step::Done;
                            Poll::Ready(Some(Err(make_callback_error(err))))
                        } else {
                            self.current_step = AsyncListV2Step::WaitForResponse {
                                task: Box::pin(async move { request.call().await }),
                            };
                            self.poll_next(cx)
                        }
                    }
                    Err(err) => {
                        self.current_step = AsyncListV2Step::Done;
                        Poll::Ready(Some(Err(err.into())))
                    }
                }
            } else {
                unreachable!()
            }
        }

pub fn credential(&self) -> &dyn CredentialProvider

获取七牛认证信息提供者

Examples found in repository ?

src/batch_operations.rs (line 214)

    fn make_request(&self) -> ApiResult<BatchOpsSyncRequestBuilder<'a, RefRegionProviderEndpoints<'a>>> {
        let request = self
            .operations
            .bucket
            .objects_manager()
            .client()
            .storage()
            .batch_ops()
            .new_request(
                RegionsProviderEndpoints::new(self.operations.bucket.region_provider()?),
                self.operations.bucket.objects_manager().credential(),
            );
        Ok(request)
    }

    fn call_request(
        &mut self,
        mut request: BatchOpsSyncRequestBuilder<'_, RefRegionProviderEndpoints>,
        request_body: RequestBody,
    ) -> ApiResult<Response<ResponseBody>> {
        self.operations
            .callbacks
            .before_request(request.parts_mut())
            .map_err(make_callback_error)?;
        let mut response_result = request.call(request_body);
        self.operations
            .callbacks
            .after_response(&mut response_result)
            .map_err(make_callback_error)?;
        response_result
    }

    fn handle_response(&mut self, response_body: ResponseBody) -> Option<ApiResult<OperationResponseData>> {
        let responses = response_body.to_operation_response_vec();
        self.buffer
            .extend(responses.into_iter().map(from_response_to_response_data_result));
        self.buffer.pop_front()
    }

    fn generate_request_body(&mut self) -> Option<RequestBody> {
        let mut request_body = RequestBody::default();
        let mut operation_count = 0usize;
        for _ in 0..self
            .operations
            .batch_size
            .as_ref()
            .map(|provider| provider.batch_size())
            .unwrap_or(DEFAULT_BATCH_SIZE)
        {
            if let Some(mut operation) = self.operations.operations.as_mut().and_then(|op| op.next()) {
                request_body = request_body.append_operations_as_str(operation.to_operation());
                operation_count += 1;
            } else {
                self.closed = true;
                break;
            }
        }
        if operation_count > 0 {
            Some(request_body)
        } else {
            None
        }
    }

    #[allow(dead_code)]
    fn assert() {
        assert_impl!(Send: Self);
        assert_impl!(Sync: Self);
    }
}

#[cfg(feature = "async")]
mod async_stream {
    use super::*;
    use futures::{future::BoxFuture, ready, FutureExt, Stream};
    use qiniu_apis::storage::batch_ops::AsyncRequestBuilder as BatchOpsAsyncRequestBuilder;
    use smart_default::SmartDefault;
    use std::{
        fmt::{self, Debug},
        io::Result as IOResult,
        pin::Pin,
        task::{Context, Poll},
    };

    /// 批量操作流
    ///
    /// 实现 [`futures::stream::Stream`] 接口，
    /// 在迭代过程中异步发起批量操作 API
    #[must_use]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    #[derive(Debug)]
    pub struct BatchOperationsStream<'a> {
        operations: BatchOperations<'a>,
        current_step: BatchOperationsStep<'a>,
        closed: bool,
    }

    #[derive(SmartDefault)]
    enum BatchOperationsStep<'a> {
        #[default]
        FromBuffer {
            buffer: VecDeque<ApiResult<OperationResponseData>>,
        },
        WaitForResponse {
            task: BoxFuture<'a, ApiResult<Response<ResponseBody>>>,
        },
        WaitForRegionProvider {
            task: BoxFuture<'a, IOResult<&'a dyn RegionsProvider>>,
        },
        Done,
    }

    impl Debug for BatchOperationsStep<'_> {
        #[inline]
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            match self {
                Self::FromBuffer { buffer } => f.debug_tuple("FromBuffer").field(buffer).finish(),
                Self::WaitForResponse { .. } => f.debug_tuple("WaitForResponse").finish(),
                Self::WaitForRegionProvider { .. } => f.debug_tuple("WaitForRegionProvider").finish(),
                Self::Done => f.debug_tuple("Done").finish(),
            }
        }
    }

    impl Stream for BatchOperationsStream<'_> {
        type Item = ApiResult<OperationResponseData>;

        #[inline]
        fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            match self.current_step {
                BatchOperationsStep::FromBuffer { .. } => self.read_from_buffer(cx),
                BatchOperationsStep::WaitForResponse { .. } => self.wait_for_response(cx),
                BatchOperationsStep::WaitForRegionProvider { .. } => self.wait_for_region(cx),
                BatchOperationsStep::Done { .. } => Poll::Ready(None),
            }
        }
    }

    impl<'a> BatchOperationsStream<'a> {
        pub(super) fn new(operations: BatchOperations<'a>) -> Self {
            Self {
                operations,
                current_step: Default::default(),
                closed: false,
            }
        }

        fn read_from_buffer(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let BatchOperationsStep::FromBuffer { buffer } = &mut self.current_step {
                if let Some(response) = buffer.pop_front() {
                    Poll::Ready(Some(response))
                } else if self.closed {
                    self.current_step = BatchOperationsStep::Done;
                    self.poll_next(cx)
                } else {
                    let bucket = self.operations.bucket;
                    self.current_step = BatchOperationsStep::WaitForRegionProvider {
                        task: Box::pin(async move { bucket.async_region_provider().await }),
                    };
                    self.poll_next(cx)
                }
            } else {
                unreachable!()
            }
        }

        fn wait_for_response(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let BatchOperationsStep::WaitForResponse { task } = &mut self.current_step {
                let mut response_result = ready!(task.poll_unpin(cx));
                if let Err(err) = self.operations.callbacks.after_response(&mut response_result) {
                    self.current_step = BatchOperationsStep::Done;
                    return Poll::Ready(Some(Err(make_callback_error(err))));
                }
                match response_result {
                    Ok(response) => {
                        self.current_step = BatchOperationsStep::FromBuffer {
                            buffer: response
                                .into_body()
                                .to_operation_response_vec()
                                .into_iter()
                                .map(from_response_to_response_data_result)
                                .collect(),
                        };
                        self.poll_next(cx)
                    }
                    Err(err) => {
                        self.current_step = BatchOperationsStep::Done;
                        Poll::Ready(Some(Err(err)))
                    }
                }
            } else {
                unreachable!()
            }
        }

        fn wait_for_region(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let BatchOperationsStep::WaitForRegionProvider { task } = &mut self.current_step {
                match ready!(task.poll_unpin(cx)) {
                    Ok(region_provider) => {
                        if let Some(request_body) = self.generate_request_body() {
                            let mut request = self.make_request(region_provider);
                            if let Err(err) = self.operations.callbacks.before_request(request.parts_mut()) {
                                self.current_step = BatchOperationsStep::Done;
                                Poll::Ready(Some(Err(make_callback_error(err))))
                            } else {
                                self.current_step = BatchOperationsStep::WaitForResponse {
                                    task: Box::pin(async move { request.call(request_body).await }),
                                };
                                self.poll_next(cx)
                            }
                        } else {
                            self.current_step = BatchOperationsStep::Done;
                            self.poll_next(cx)
                        }
                    }
                    Err(err) => {
                        self.current_step = BatchOperationsStep::Done;
                        Poll::Ready(Some(Err(err.into())))
                    }
                }
            } else {
                unreachable!()
            }
        }

        fn generate_request_body(&mut self) -> Option<RequestBody> {
            let mut request_body = RequestBody::default();
            let mut operation_count = 0usize;
            for _ in 0..self
                .operations
                .batch_size
                .as_ref()
                .map(|provider| provider.batch_size())
                .unwrap_or(DEFAULT_BATCH_SIZE)
            {
                if let Some(mut operation) = self.operations.operations.as_mut().and_then(|op| op.next()) {
                    request_body = request_body.append_operations_as_str(operation.to_operation());
                    operation_count += 1;
                } else {
                    self.closed = true;
                    break;
                }
            }
            if operation_count > 0 {
                Some(request_body)
            } else {
                None
            }
        }

        fn make_request(
            &self,
            region_provider: &'a dyn RegionsProvider,
        ) -> BatchOpsAsyncRequestBuilder<'a, RefRegionProviderEndpoints<'a>> {
            self.operations
                .bucket
                .objects_manager()
                .client()
                .storage()
                .batch_ops()
                .new_async_request(
                    RegionsProviderEndpoints::new(region_provider),
                    self.operations.bucket.objects_manager().credential(),
                )
        }

More examples

Hide additional examples

src/list.rs (line 315)

        fn v1_make_request<'a>(
            params: &mut ListParams<'a>,
        ) -> ApiResult<GetObjectsV1SyncRequestBuilder<'a, RefRegionProviderEndpoints<'a>>> {
            let mut request = params
                .bucket
                .objects_manager()
                .client()
                .storage()
                .get_objects()
                .new_request(
                    RegionsProviderEndpoints::new(params.bucket.region_provider()?),
                    params.bucket.objects_manager().credential(),
                );
            request.query_pairs(params.to_query_params());
            Ok(request)
        }

        fn v1_call_request(
            mut request: GetObjectsV1SyncRequestBuilder<'_, RefRegionProviderEndpoints>,
            callbacks: &mut Callbacks<'_>,
        ) -> ApiResult<Response<GetObjectsV1ResponseBody>> {
            callbacks
                .before_request(request.parts_mut())
                .map_err(make_callback_error)?;
            let mut response_result = request.call();
            callbacks
                .after_response(&mut response_result)
                .map_err(make_callback_error)?;
            response_result
        }

        fn v1_handle_response(
            body: GetObjectsV1ResponseBody,
            params: &mut ListParams<'_>,
            buffer: &mut VecDeque<ListedObjectEntry>,
        ) {
            params.marker.set(body.get_marker_as_str());
            let listed_object_entries = body.get_items().to_listed_object_entry_vec();
            params.limit.saturating_decrease(listed_object_entries.len());
            *buffer = listed_object_entries.into();
        }

        fn v2_next(
            params: &mut ListParams<'_>,
            callbacks: &mut Callbacks<'_>,
            step: &mut SyncListV2Step,
        ) -> Option<ApiResult<ListedObjectEntry>> {
            match step {
                SyncListV2Step::Start => match v2_call(params, callbacks) {
                    Ok(Some(mut lines)) => v2_read_entry_from_lines(params, &mut lines).tap_some(|result| {
                        if result.is_ok() {
                            *step = SyncListV2Step::Lines { lines };
                        } else {
                            *step = SyncListV2Step::Done;
                        }
                    }),
                    Ok(None) => {
                        *step = SyncListV2Step::Done;
                        None
                    }
                    Err(err) => {
                        *step = SyncListV2Step::Done;
                        Some(Err(err))
                    }
                },
                SyncListV2Step::Lines { lines } => match v2_read_entry_from_lines(params, lines) {
                    Some(Ok(entry)) => Some(Ok(entry)),
                    Some(Err(err)) => {
                        warn!("Read Error from ListV2 Response Body: {}", err);
                        *step = SyncListV2Step::Start;
                        v2_next(params, callbacks, step)
                    }
                    None => {
                        *step = SyncListV2Step::Start;
                        v2_next(params, callbacks, step)
                    }
                },
                SyncListV2Step::Done => None,
            }
        }

        fn v2_read_entry_from_lines(
            params: &mut ListParams<'_>,
            lines: &mut Lines<BufReader<SyncResponseBody>>,
        ) -> Option<ApiResult<ListedObjectEntry>> {
            if params.limit.exhausted() {
                return None;
            }
            loop {
                match lines.next() {
                    Some(Ok(line)) if line.is_empty() => {
                        continue;
                    }
                    Some(Ok(line)) => match serde_json::from_str::<ListedObjectEntryV2>(&line) {
                        Ok(parsed) => {
                            params.marker.set(parsed.marker.as_deref());
                            if let Some(item) = parsed.item {
                                params.limit.saturating_decrease(1);
                                return Some(Ok(item));
                            } else {
                                continue;
                            }
                        }
                        Err(err) => {
                            return Some(Err(err.into()));
                        }
                    },
                    Some(Err(err)) => {
                        return Some(Err(err.into()));
                    }
                    None => {
                        return None;
                    }
                }
            }
        }

        fn v2_call(
            params: &mut ListParams<'_>,
            callbacks: &mut Callbacks<'_>,
        ) -> ApiResult<Option<Lines<BufReader<SyncResponseBody>>>> {
            if params.have_done() {
                return Ok(None);
            }
            let request = v2_make_request(params)?;
            let response_result = v2_call_request(request, callbacks);
            Ok(Some(BufReader::new(response_result?.into_body()).lines()))
        }

        fn v2_make_request<'a>(
            params: &mut ListParams<'a>,
        ) -> ApiResult<GetObjectsV2SyncRequestBuilder<'a, RefRegionProviderEndpoints<'a>>> {
            let mut request = params
                .bucket
                .objects_manager()
                .client()
                .storage()
                .get_objects_v2()
                .new_request(
                    RegionsProviderEndpoints::new(params.bucket.region_provider()?),
                    params.bucket.objects_manager().credential(),
                );
            request.query_pairs(params.to_query_params());
            Ok(request)
        }

        fn v2_call_request(
            mut request: GetObjectsV2SyncRequestBuilder<'_, RefRegionProviderEndpoints>,
            callbacks: &mut Callbacks<'_>,
        ) -> ApiResult<Response<SyncResponseBody>> {
            callbacks
                .before_request(request.parts_mut())
                .map_err(make_callback_error)?;
            let mut response_result = request.call();
            callbacks
                .after_response(&mut response_result)
                .map_err(make_callback_error)?;
            response_result
        }
    }
}

#[cfg(feature = "async")]
mod async_list_stream {
    use super::*;
    use futures::{future::BoxFuture, io::Lines as AsyncLines, ready, AsyncBufReadExt, FutureExt, Stream, StreamExt};
    use std::{
        fmt::{self, Debug},
        io::Result as IOResult,
        pin::Pin,
        task::{Context, Poll},
    };

    enum AsyncListVersionWithStep<'a> {
        V1(AsyncListV1Step<'a>),
        V2(AsyncListV2Step<'a>),
    }

    impl From<ListVersion> for AsyncListVersionWithStep<'_> {
        fn from(version: ListVersion) -> Self {
            match version {
                ListVersion::V1 => Self::V1(Default::default()),
                ListVersion::V2 => Self::V2(Default::default()),
            }
        }
    }

    #[derive(SmartDefault)]
    enum AsyncListV1Step<'a> {
        #[default]
        FromBuffer {
            buffer: VecDeque<ListedObjectEntry>,
        },
        WaitForResponse {
            task: BoxFuture<'a, ApiResult<Response<GetObjectsV1ResponseBody>>>,
        },
        WaitForRegionProvider {
            task: BoxFuture<'a, IOResult<&'a dyn RegionsProvider>>,
        },
        Done,
    }

    impl Debug for AsyncListV1Step<'_> {
        #[inline]
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            match self {
                Self::FromBuffer { buffer } => f.debug_tuple("FromBuffer").field(buffer).finish(),
                Self::WaitForResponse { .. } => f.debug_tuple("WaitForResponse").finish(),
                Self::WaitForRegionProvider { .. } => f.debug_tuple("WaitForRegionProvider").finish(),
                Self::Done => f.debug_tuple("Done").finish(),
            }
        }
    }

    trait StreamWithMarker: Stream {
        fn marker(&self) -> Option<&str>;
    }
    type BoxStreamWithMarker<'a, T> = Pin<Box<dyn StreamWithMarker<Item = T> + Send + 'a>>;
    type ListedObjectEntryResultStream<'a> = BoxStreamWithMarker<'a, ApiResult<ListedObjectEntry>>;

    /// 对象列举流
    ///
    /// 实现 [`futures::stream::Stream`] 接口，
    /// 在迭代过程中异步发起 API 列举对象信息
    ///
    /// 可以通过 [`crate::ListBuilder::stream`] 方法获取该迭代器。
    #[must_use]
    #[cfg_attr(feature = "docs", doc(cfg(feature = "async")))]
    pub struct ListStream<'a>(ListedObjectEntryResultStream<'a>);

    impl<'a> ListStream<'a> {
        pub(in super::super) fn new(
            bucket: &'a Bucket,
            limit: Option<usize>,
            prefix: Option<Cow<'a, str>>,
            marker: Option<Cow<'a, str>>,
            need_parts: bool,
            version: ListVersion,
            callbacks: Callbacks<'a>,
        ) -> Self {
            Self(match version {
                ListVersion::V1 => v1_next(bucket, limit, prefix, marker, need_parts, callbacks),
                ListVersion::V2 => v2_next(bucket, limit, prefix, marker, need_parts, callbacks),
            })
        }

        #[allow(dead_code)]
        fn assert() {
            assert_impl!(Send: Self);
            // assert_impl!(Sync: Self);
        }

        /// 获取上一次列举返回的位置标记
        pub fn marker(&self) -> Option<&str> {
            self.0.marker()
        }
    }

    impl Stream for ListStream<'_> {
        type Item = ApiResult<ListedObjectEntry>;

        #[inline]
        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            self.0.poll_next_unpin(cx)
        }
    }

    impl Debug for ListStream<'_> {
        #[inline]
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.debug_struct("ListStream").finish()
        }
    }

    #[derive(Debug)]
    struct ListV1Stream<'a> {
        params: ListParams<'a>,
        callbacks: Callbacks<'a>,
        current_step: AsyncListV1Step<'a>,
    }

    fn v1_next<'a>(
        bucket: &'a Bucket,
        limit: Option<usize>,
        prefix: Option<Cow<'a, str>>,
        marker: Option<Cow<'a, str>>,
        need_parts: bool,
        callbacks: Callbacks<'a>,
    ) -> ListedObjectEntryResultStream<'a> {
        let params = ListParams {
            bucket,
            prefix,
            need_parts,
            limit: Limit::new(limit, ListVersion::V1),
            marker: Marker::new(marker),
        };
        Box::pin(ListV1Stream {
            params,
            callbacks,
            current_step: Default::default(),
        })
    }

    impl Stream for ListV1Stream<'_> {
        type Item = ApiResult<ListedObjectEntry>;

        #[inline]
        fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            match self.current_step {
                AsyncListV1Step::FromBuffer { .. } => self.read_from_buffer(cx),
                AsyncListV1Step::WaitForResponse { .. } => self.wait_for_response(cx),
                AsyncListV1Step::WaitForRegionProvider { .. } => self.wait_for_region(cx),
                AsyncListV1Step::Done => Poll::Ready(None),
            }
        }
    }

    impl StreamWithMarker for ListV1Stream<'_> {
        fn marker(&self) -> Option<&str> {
            self.params.marker.as_ref()
        }
    }

    impl ListV1Stream<'_> {
        fn read_from_buffer(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV1Step::FromBuffer { buffer } = &mut self.current_step {
                if let Some(object) = buffer.pop_front() {
                    Poll::Ready(Some(Ok(object)))
                } else {
                    if self.params.have_done() {
                        self.current_step = AsyncListV1Step::Done;
                    } else {
                        let bucket = self.params.bucket;
                        self.current_step = AsyncListV1Step::WaitForRegionProvider {
                            task: Box::pin(async move { bucket.async_region_provider().await }),
                        };
                    }
                    self.poll_next(cx)
                }
            } else {
                unreachable!()
            }
        }

        fn wait_for_region(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV1Step::WaitForRegionProvider { task } = &mut self.current_step {
                match ready!(task.poll_unpin(cx)) {
                    Ok(region_provider) => {
                        let credential = self.params.bucket.objects_manager().credential();
                        let mut request = self
                            .params
                            .bucket
                            .objects_manager()
                            .client()
                            .storage()
                            .get_objects()
                            .new_async_request(RegionsProviderEndpoints::new(region_provider), credential);
                        request.query_pairs(self.params.to_query_params());
                        if let Err(err) = self.callbacks.before_request(request.parts_mut()) {
                            self.current_step = AsyncListV1Step::Done;
                            Poll::Ready(Some(Err(make_callback_error(err))))
                        } else {
                            self.current_step = AsyncListV1Step::WaitForResponse {
                                task: Box::pin(async move { request.call().await }),
                            };
                            self.poll_next(cx)
                        }
                    }
                    Err(err) => {
                        self.current_step = AsyncListV1Step::Done;
                        Poll::Ready(Some(Err(err.into())))
                    }
                }
            } else {
                unreachable!()
            }
        }

        fn wait_for_response(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV1Step::WaitForResponse { task } = &mut self.current_step {
                let mut response_result = ready!(task.poll_unpin(cx));
                if let Err(err) = self.callbacks.after_response(&mut response_result) {
                    self.current_step = AsyncListV1Step::Done;
                    Poll::Ready(Some(Err(make_callback_error(err))))
                } else {
                    match response_result {
                        Ok(response) => {
                            let body = response.into_body();
                            let listed_object_entries = body.get_items().to_listed_object_entry_vec();
                            self.params.marker.set(body.get_marker_as_str());
                            self.params.limit.saturating_decrease(listed_object_entries.len());
                            self.current_step = AsyncListV1Step::FromBuffer {
                                buffer: listed_object_entries.into(),
                            };
                            self.poll_next(cx)
                        }
                        Err(err) => {
                            self.current_step = AsyncListV1Step::Done;
                            Poll::Ready(Some(Err(err)))
                        }
                    }
                }
            } else {
                unreachable!()
            }
        }
    }

    #[derive(SmartDefault)]
    enum AsyncListV2Step<'a> {
        #[default]
        Start,
        WaitForRegionProvider {
            task: BoxFuture<'a, IOResult<&'a dyn RegionsProvider>>,
        },
        WaitForResponse {
            task: BoxFuture<'a, ApiResult<Response<AsyncResponseBody>>>,
        },
        WaitForEntries {
            lines: AsyncLines<AsyncBufReader<AsyncResponseBody>>,
            empty: bool,
        },
        Done,
    }

    struct ListV2Stream<'a> {
        params: ListParams<'a>,
        callbacks: Callbacks<'a>,
        current_step: AsyncListV2Step<'a>,
    }

    #[allow(clippy::too_many_arguments)]
    fn v2_next<'a>(
        bucket: &'a Bucket,
        limit: Option<usize>,
        prefix: Option<Cow<'a, str>>,
        marker: Option<Cow<'a, str>>,
        need_parts: bool,
        callbacks: Callbacks<'a>,
    ) -> ListedObjectEntryResultStream<'a> {
        let params = ListParams {
            bucket,
            prefix,
            need_parts,
            limit: Limit::new(limit, ListVersion::V2),
            marker: Marker::new(marker),
        };
        Box::pin(ListV2Stream {
            params,
            callbacks,
            current_step: Default::default(),
        })
    }

    impl Stream for ListV2Stream<'_> {
        type Item = ApiResult<ListedObjectEntry>;

        #[inline]
        fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            match self.current_step {
                AsyncListV2Step::Start { .. } => self.start(cx),
                AsyncListV2Step::WaitForResponse { .. } => self.wait_for_response(cx),
                AsyncListV2Step::WaitForRegionProvider { .. } => self.wait_for_region(cx),
                AsyncListV2Step::WaitForEntries { .. } => self.wait_for_entries(cx),
                AsyncListV2Step::Done => Poll::Ready(None),
            }
        }
    }

    impl StreamWithMarker for ListV2Stream<'_> {
        fn marker(&self) -> Option<&str> {
            self.params.marker.as_ref()
        }
    }

    impl ListV2Stream<'_> {
        fn start(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV2Step::Start { .. } = &mut self.current_step {
                if self.params.have_done() {
                    self.current_step = AsyncListV2Step::Done;
                } else {
                    let bucket = self.params.bucket;
                    self.current_step = AsyncListV2Step::WaitForRegionProvider {
                        task: Box::pin(async move { bucket.async_region_provider().await }),
                    };
                }
                self.poll_next(cx)
            } else {
                unreachable!()
            }
        }

        fn wait_for_region(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<<Self as Stream>::Item>> {
            if let AsyncListV2Step::WaitForRegionProvider { task } = &mut self.current_step {
                match ready!(task.poll_unpin(cx)) {
                    Ok(region_provider) => {
                        let credential = self.params.bucket.objects_manager().credential();
                        let mut request = self
                            .params
                            .bucket
                            .objects_manager()
                            .client()
                            .storage()
                            .get_objects_v2()
                            .new_async_request(RegionsProviderEndpoints::new(region_provider), credential);
                        request.query_pairs(self.params.to_query_params());
                        if let Err(err) = self.callbacks.before_request(request.parts_mut()) {
                            self.current_step = AsyncListV2Step::Done;
                            Poll::Ready(Some(Err(make_callback_error(err))))
                        } else {
                            self.current_step = AsyncListV2Step::WaitForResponse {
                                task: Box::pin(async move { request.call().await }),
                            };
                            self.poll_next(cx)
                        }
                    }
                    Err(err) => {
                        self.current_step = AsyncListV2Step::Done;
                        Poll::Ready(Some(Err(err.into())))
                    }
                }
            } else {
                unreachable!()
            }
        }

src/bucket.rs (line 521)

    pub(super) fn region_provider(&self) -> IOResult<&dyn RegionsProvider> {
        self.0
            .region_provider
            .as_ref()
            .map(|r| Ok(r as &dyn RegionsProvider))
            .unwrap_or_else(|| {
                self.0
                    .bucket_regions_provider
                    .get_or_try_init(|| {
                        Ok(self.0.objects_manager.queryer().query(
                            self.0
                                .objects_manager
                                .credential()
                                .get(Default::default())?
                                .access_key()
                                .to_owned(),
                            self.name().to_owned(),
                        ))
                    })
                    .map(|r| r as &dyn RegionsProvider)
            })
    }

    #[cfg(feature = "async")]
    pub(super) async fn async_region_provider(&self) -> IOResult<&dyn RegionsProvider> {
        return if let Some(region_provider) = self.0.region_provider.as_ref() {
            Ok(region_provider)
        } else {
            self.0
                .async_bucket_regions_provider
                .get_or_try_init(create_region_provider(&self.0.objects_manager, self.name()))
                .await
                .map(|r| r as &dyn RegionsProvider)
        };

        async fn create_region_provider(
            objects_manager: &ObjectsManager,
            bucket_name: &BucketName,
        ) -> IOResult<BucketRegionsProvider> {
            Ok(objects_manager.queryer().query(
                objects_manager
                    .credential()
                    .async_get(Default::default())
                    .await?
                    .access_key()
                    .to_owned(),
                bucket_name.to_owned(),
            ))
        }

pub fn queryer(&self) -> &BucketRegionsQueryer

获取七牛存储空间相关区域查询器

Examples found in repository ?

src/bucket.rs (line 518)

    pub(super) fn region_provider(&self) -> IOResult<&dyn RegionsProvider> {
        self.0
            .region_provider
            .as_ref()
            .map(|r| Ok(r as &dyn RegionsProvider))
            .unwrap_or_else(|| {
                self.0
                    .bucket_regions_provider
                    .get_or_try_init(|| {
                        Ok(self.0.objects_manager.queryer().query(
                            self.0
                                .objects_manager
                                .credential()
                                .get(Default::default())?
                                .access_key()
                                .to_owned(),
                            self.name().to_owned(),
                        ))
                    })
                    .map(|r| r as &dyn RegionsProvider)
            })
    }

    #[cfg(feature = "async")]
    pub(super) async fn async_region_provider(&self) -> IOResult<&dyn RegionsProvider> {
        return if let Some(region_provider) = self.0.region_provider.as_ref() {
            Ok(region_provider)
        } else {
            self.0
                .async_bucket_regions_provider
                .get_or_try_init(create_region_provider(&self.0.objects_manager, self.name()))
                .await
                .map(|r| r as &dyn RegionsProvider)
        };

        async fn create_region_provider(
            objects_manager: &ObjectsManager,
            bucket_name: &BucketName,
        ) -> IOResult<BucketRegionsProvider> {
            Ok(objects_manager.queryer().query(
                objects_manager
                    .credential()
                    .async_get(Default::default())
                    .await?
                    .access_key()
                    .to_owned(),
                bucket_name.to_owned(),
            ))
        }

pub fn bucket(&self, name: impl Into<BucketName>) -> Bucket

获取七牛存储空间管理器

pub fn bucket_with_region(
 &self,
 name: impl Into<BucketName>,
 region_provider: impl RegionsProvider + 'static
) -> Bucket

获取七牛存储空间管理器

可以提供区域信息提供者

Trait Implementations§

impl Clone for ObjectsManager

fn clone(&self) -> ObjectsManager

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 ObjectsManager

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

Formats the value using the given formatter. Read more

Auto Trait Implementations§

impl !UnwindSafe for ObjectsManager

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