ocpi_kit/client/
paging.rs

1//! Crawling a paginated list endpoint, with the concurrency correction the spec asks for.
2
3use std::collections::VecDeque;
4
5use http::Method;
6use serde::de::DeserializeOwned;
7
8use crate::ModuleId;
9use crate::convert::wire::ObjectKind;
10use crate::transport::{CrawlAdjustment, OcpiError, OcpiRequest, PageMeta, RoutingHeaders, crawl_adjustment};
11use crate::types::Url;
12
13use super::http::Transport;
14use super::peer::Peer;
15
16/// An asynchronous crawl over every page of a list endpoint.
17///
18/// Following the `Link: <…>; rel="next"` header is only most of the job. The specification also
19/// describes what to do when the result set changes underneath the crawl:
20///
21/// > *While a client crawls over the pages … a new object might be created on the server. The
22/// > client detects this: the `X-Total-Count` will be higher on the next call. Even so, the client
23/// > does not have to retry any requests when this happens because only the last page will be
24/// > different.*
25///
26/// > *When there are for example 1000 objects matching a query … while crawling over the pages one
27/// > of these objects is updated. The client detects this: `X-Total-Count` will be lower in the
28/// > next request. It is advised to redo the previous GET with the `offset` lowered by 1 (if the
29/// > `offset` was not 0) and after that continue crawling the 'next' page links.*
30///
31/// [`PageStream`] does both, and reports the correction it made through
32/// [`PageStream::corrections`] so a pull that keeps shifting is visible rather than silent.
33///
34/// ```no_run
35/// # use ocpi_kit::client::PageStream;
36/// # use ocpi_kit::v2_3_0::locations::Location;
37/// # async fn crawl(mut stream: PageStream<'_, Location>) -> Result<(), Box<dyn std::error::Error>> {
38/// while let Some(location) = stream.next().await? {
39///     println!("{}", location.id);
40/// }
41/// println!("{} objects over {} pages", stream.seen(), stream.pages_fetched());
42/// # Ok(())
43/// # }
44/// ```
45///
46/// Spec: 2.3.0 §transport_and_format_paginated_response
47pub struct PageStream<'a, T> {
48    transport: &'a Transport,
49    peer: &'a Peer,
50    module: ModuleId,
51    routing: RoutingHeaders,
52    next: Option<Url>,
53    buffer: VecDeque<T>,
54    last_total: Option<u64>,
55    last_offset: u64,
56    pages: usize,
57    seen: usize,
58    corrections: usize,
59    max_pages: usize,
60    bridge: Option<ObjectKind>,
61}
62
63/// The number of pages a crawl will fetch before giving up, unless configured otherwise.
64///
65/// A peer that answers every page with a `Link` to itself would otherwise loop forever.
66pub const DEFAULT_MAX_PAGES: usize = 10_000;
67
68impl<'a, T: DeserializeOwned> PageStream<'a, T> {
69    /// Starts a crawl at `first`.
70    #[must_use]
71    pub fn new(
72        transport: &'a Transport,
73        peer: &'a Peer,
74        module: ModuleId,
75        routing: RoutingHeaders,
76        first: Url,
77    ) -> Self {
78        Self {
79            transport,
80            peer,
81            module,
82            routing,
83            next: Some(first),
84            buffer: VecDeque::new(),
85            last_total: None,
86            last_offset: 0,
87            pages: 0,
88            seen: 0,
89            corrections: 0,
90            max_pages: DEFAULT_MAX_PAGES,
91            bridge: None,
92        }
93    }
94
95    /// Translates every page out of the peer's OCPI version into the canonical model.
96    ///
97    /// `kind` says which object the endpoint carries. A peer that already speaks the canonical
98    /// version costs nothing: the translation is skipped, not applied as an identity.
99    #[must_use]
100    pub const fn bridging(mut self, kind: ObjectKind) -> Self {
101        self.bridge = Some(kind);
102        self
103    }
104
105    /// Caps how many pages this crawl will fetch.
106    #[must_use]
107    pub const fn with_max_pages(mut self, max_pages: usize) -> Self {
108        self.max_pages = max_pages;
109        self
110    }
111
112    /// The next object, fetching another page when the buffer runs dry.
113    ///
114    /// # Errors
115    ///
116    /// Propagates transport, decoding and OCPI-level errors from the page fetch.
117    pub async fn next(&mut self) -> Result<Option<T>, OcpiError> {
118        loop {
119            if let Some(item) = self.buffer.pop_front() {
120                self.seen += 1;
121                return Ok(Some(item));
122            }
123            let Some(url) = self.next.take() else { return Ok(None) };
124            if self.pages >= self.max_pages {
125                return Err(OcpiError::Transport(format!(
126                    "pagination did not terminate after {} pages; the peer keeps returning a \
127                     `Link` header",
128                    self.max_pages
129                )));
130            }
131            self.fetch(url).await?;
132        }
133    }
134
135    /// Collects the whole list.
136    ///
137    /// # Errors
138    ///
139    /// Propagates any error from [`PageStream::next`].
140    pub async fn collect_all(mut self) -> Result<Vec<T>, OcpiError> {
141        let mut out = Vec::new();
142        while let Some(item) = self.next().await? {
143            out.push(item);
144        }
145        Ok(out)
146    }
147
148    /// How many pages have been fetched.
149    #[must_use]
150    pub const fn pages_fetched(&self) -> usize {
151        self.pages
152    }
153
154    /// How many objects have been yielded.
155    #[must_use]
156    pub const fn seen(&self) -> usize {
157        self.seen
158    }
159
160    /// How many times the crawl was rewound because `X-Total-Count` shrank.
161    ///
162    /// A non-zero count means objects were changing while the crawl ran.
163    #[must_use]
164    pub const fn corrections(&self) -> usize {
165        self.corrections
166    }
167
168    /// The total the peer reported for the query, from the most recent page.
169    #[must_use]
170    pub const fn total_count(&self) -> Option<u64> {
171        self.last_total
172    }
173
174    async fn fetch(&mut self, url: Url) -> Result<(), OcpiError> {
175        let offset = offset_of(&url);
176        let request =
177            OcpiRequest::new(Method::GET, url.clone(), self.module.clone()).routed(self.routing.clone());
178        let page = self.page(&request).await?;
179        self.pages += 1;
180
181        // The correction is measured against *this* request's offset, not the previous page's:
182        // the spec's "redo the previous GET with the offset lowered by 1" is about the GET that
183        // detected the shrink. Rewinding to the page before it would re-yield a whole page of
184        // objects the crawl has already handed out.
185        match crawl_adjustment(self.last_total, page.meta.total_count, offset) {
186            CrawlAdjustment::RefetchAt(new_offset) => {
187                self.corrections += 1;
188                tracing::debug!(
189                    previous_total = self.last_total,
190                    new_total = page.meta.total_count,
191                    new_offset,
192                    "X-Total-Count shrank mid-crawl; rewinding one object as the spec advises",
193                );
194                self.last_total = page.meta.total_count;
195                self.next = Some(with_offset(&url, new_offset));
196                Ok(())
197            }
198            CrawlAdjustment::Continue => {
199                self.last_total = page.meta.total_count;
200                self.last_offset = offset;
201                self.buffer.extend(page.items);
202                self.next = next_url(&page.meta);
203                Ok(())
204            }
205        }
206    }
207
208    /// One page, translated out of the peer's version when the crawl was asked to.
209    async fn page(&self, request: &OcpiRequest) -> Result<crate::transport::Page<T>, OcpiError> {
210        let theirs = self.peer.version();
211        let Some(kind) = self.bridge.filter(|_| *theirs != crate::CANONICAL_VERSION) else {
212            return self.transport.send_page::<T>(request, self.peer.token(), self.peer.quirks()).await;
213        };
214        let page = self
215            .transport
216            .send_page::<serde_json::Value>(request, self.peer.token(), self.peer.quirks())
217            .await?;
218        let converted = kind
219            .bridge(theirs, &crate::CANONICAL_VERSION, serde_json::Value::Array(page.items))
220            .map_err(|e| OcpiError::Decode { path: "/".to_owned(), message: e.to_string() })?;
221        let items = serde_path_to_error::deserialize(converted.value).map_err(|e| OcpiError::Decode {
222            path: e.path().to_string(),
223            message: e.into_inner().to_string(),
224        })?;
225        Ok(crate::transport::Page { items, meta: page.meta })
226    }
227}
228
229fn next_url(meta: &PageMeta) -> Option<Url> {
230    meta.next.clone()
231}
232
233/// Reads the `offset` query parameter of a page URL, defaulting to 0.
234fn offset_of(url: &Url) -> u64 {
235    url.as_str()
236        .split_once('?')
237        .map(|(_, query)| query)
238        .and_then(|query| {
239            query.split('&').find_map(|pair| {
240                let (key, value) = pair.split_once('=')?;
241                (key == "offset").then(|| value.parse().ok())?
242            })
243        })
244        .unwrap_or(0)
245}
246
247/// Replaces or adds the `offset` query parameter.
248fn with_offset(url: &Url, offset: u64) -> Url {
249    let (base, query) = match url.as_str().split_once('?') {
250        Some((base, query)) => (base, Some(query)),
251        None => (url.as_str(), None),
252    };
253    let mut parts: Vec<String> = query
254        .map(|q| q.split('&').filter(|pair| !pair.starts_with("offset=")).map(ToOwned::to_owned).collect())
255        .unwrap_or_default();
256    parts.insert(0, format!("offset={offset}"));
257    Url::new_lenient(format!("{base}?{}", parts.join("&")))
258}
259
260#[cfg(test)]
261mod tests {
262    use super::*;
263
264    #[test]
265    fn the_offset_is_read_from_and_written_to_the_query() {
266        let url = Url::new("https://e.com/cdrs?offset=150&limit=50").unwrap();
267        assert_eq!(offset_of(&url), 150);
268        assert_eq!(with_offset(&url, 149).as_str(), "https://e.com/cdrs?offset=149&limit=50");
269
270        let bare = Url::new("https://e.com/cdrs").unwrap();
271        assert_eq!(offset_of(&bare), 0);
272        assert_eq!(with_offset(&bare, 10).as_str(), "https://e.com/cdrs?offset=10");
273    }
274
275    #[test]
276    fn other_filters_survive_a_rewind() {
277        // "The Link should also contain any filters present in the original request."
278        let url =
279            Url::new("https://e.com/cdrs?offset=100&limit=100&date_from=2016-01-01T00%3A00%3A00Z").unwrap();
280        assert_eq!(
281            with_offset(&url, 99).as_str(),
282            "https://e.com/cdrs?offset=99&limit=100&date_from=2016-01-01T00%3A00%3A00Z"
283        );
284    }
285}
ocpi_kit/client/paging.rs

ocpi_kit/client/
paging.rs
