spiral/
pagination.rs

1//! Consistent pagination for resource collections.
2use std::collections::HashSet;
3use std::pin::Pin;
4use std::sync::Arc;
5
6use futures::future::BoxFuture;
7use futures::{Stream, StreamExt, TryStreamExt, stream};
8
9use crate::Error;
10use crate::http::continuation;
11
12/// A lazy stream of collection items, fetching the next page only when needed.
13pub type ListStream<'a, T> = Pin<Box<dyn Stream<Item = Result<T, Error>> + Send + 'a>>;
14
15/// One response from a resource collection. Tokens are opaque and retain the request's filters.
16#[derive(Debug, Clone, serde::Deserialize)]
17pub struct Page<T> {
18    /// Items returned by this request.
19    pub items: Vec<T>,
20    /// Pass to `List::page_token` to resume the same query.
21    pub next_page_token: Option<String>,
22}
23
24#[derive(Clone, Default)]
25pub(crate) struct PageOptions {
26    pub size: Option<u32>,
27    pub token: Option<String>,
28}
29
30type Fetch<'a, T> =
31    Arc<dyn Fn(PageOptions) -> BoxFuture<'a, Result<Page<T>, Error>> + Send + Sync + 'a>;
32
33/// A lazy collection request. Use `send` for one page or `stream` to follow every page.
34/// Constructing or configuring a request performs no I/O. Each page resolves current credentials.
35pub struct List<'a, T> {
36    fetch: Fetch<'a, T>,
37    options: PageOptions,
38}
39impl<'a, T: Send + 'a> List<'a, T> {
40    pub(crate) fn new(
41        fetch: impl Fn(PageOptions) -> BoxFuture<'a, Result<Page<T>, Error>> + Send + Sync + 'a,
42    ) -> Self {
43        Self {
44            fetch: Arc::new(fetch),
45            options: PageOptions::default(),
46        }
47    }
48    pub(crate) fn with_options(mut self, options: PageOptions) -> Self {
49        self.options = options;
50        self
51    }
52    /// Set the page size. Zero is rejected; the service enforces its upper limit.
53    pub fn page_size(mut self, size: u32) -> Self {
54        self.options.size = Some(size);
55        self
56    }
57    /// Resume at an opaque token from a previous response to the same query.
58    pub fn page_token(mut self, token: impl Into<String>) -> Self {
59        self.options.token = Some(token.into());
60        self
61    }
62    /// Fetch exactly one page, preserving an explicit continuation token in the result.
63    pub async fn send(self) -> Result<Page<T>, Error> {
64        if self.options.size == Some(0) || self.options.token.as_deref() == Some("") {
65            return Err(Error::invalid(
66                "page size and continuation token must be nonempty",
67            ));
68        }
69        (self.fetch)(self.options).await
70    }
71    /// Stream items with bounded page buffering, stopping at the first error.
72    /// Empty intermediate pages are followed; repeated tokens are rejected.
73    pub fn stream(self) -> ListStream<'a, T> {
74        let mut seen = HashSet::new();
75        if let Some(token) = &self.options.token {
76            seen.insert(token.clone());
77        }
78        stream::try_unfold((Some(self), seen), |(request, mut seen)| async move {
79            let Some(request) = request else {
80                return Ok::<_, Error>(None);
81            };
82            let fetch = Arc::clone(&request.fetch);
83            let mut options = request.options.clone();
84            let page = request.send().await?;
85            options.token = continuation(page.next_page_token, &mut seen)?;
86            let next = options.token.as_ref().map(|_| Self {
87                fetch,
88                options: options.clone(),
89            });
90            Ok(Some((
91                stream::iter(page.items.into_iter().map(Ok)),
92                (next, seen),
93            )))
94        })
95        .try_flatten()
96        .boxed()
97    }
98}
spiral/pagination.rs

spiral/
pagination.rs
