ably_chat/
pagination.rs

1//! Cursor pagination as `Page<T>` + `Stream` over RFC 5988 `Link` headers
2//! (ADR-0009). The dispatch layer surfaces the response headers the generated
3//! functions discard; this module parses the `rel="next"` cursor and follows it.
4
5use futures::Stream;
6use reqwest::Method;
7use reqwest::header::{HeaderMap, LINK};
8use serde::de::DeserializeOwned;
9
10use crate::client::Client;
11use crate::dispatch::{RawResponse, decode_json};
12use crate::error::Result;
13
14/// One page of a paginated collection: the decoded items plus the parsed
15/// `rel="next"` cursor (ADR-0009).
16///
17/// Obtain the next page manually with [`next`](Self::next), or follow the whole
18/// chain lazily with [`into_stream`](Self::into_stream).
19#[derive(Clone, Debug)]
20pub struct Page<T> {
21    items: Vec<T>,
22    /// Absolute URL of the next page, if any (already resolved against the base).
23    next: Option<String>,
24    client: Client,
25}
26
27impl<T> Page<T> {
28    /// The items on this page.
29    pub fn items(&self) -> &[T] {
30        &self.items
31    }
32
33    /// Consumes the page, returning its items.
34    pub fn into_items(self) -> Vec<T> {
35        self.items
36    }
37
38    /// Whether a further page is available via [`next`](Self::next).
39    pub fn has_next(&self) -> bool {
40        self.next.is_some()
41    }
42}
43
44impl<T: DeserializeOwned + Send + 'static> Page<T> {
45    /// Fetches the first page for a base-relative `path` and `query`. The shared
46    /// entry point for both the `.await` (single page) and `.into_stream()`
47    /// forms of the paginated builders, so the query is built in exactly one
48    /// place.
49    pub(crate) async fn fetch_first(
50        client: Client,
51        path: String,
52        query: Vec<(&'static str, String)>,
53    ) -> Result<Self> {
54        let resp = client
55            .inner
56            .send(Method::GET, &path, &query, None, false)
57            .await?;
58        Self::from_response(client, &resp)
59    }
60
61    /// Builds a page from a raw response, decoding the JSON array body and
62    /// resolving the `next` link (if present) to an absolute URL.
63    fn from_response(client: Client, resp: &RawResponse) -> Result<Self> {
64        let items = decode_json::<Vec<T>>(&resp.body)?;
65        let next =
66            parse_next_link(&resp.headers).map(|link| resolve_url(&client.inner.base, &link));
67        Ok(Self {
68            items,
69            next,
70            client,
71        })
72    }
73
74    /// Fetches the next page, or `None` when the current page is the last.
75    pub async fn next(&self) -> Result<Option<Page<T>>> {
76        match &self.next {
77            None => Ok(None),
78            Some(url) => {
79                let resp = self
80                    .client
81                    .inner
82                    .send_url(Method::GET, url.clone(), &[], None, false)
83                    .await?;
84                Ok(Some(Self::from_response(self.client.clone(), &resp)?))
85            }
86        }
87    }
88
89    /// Streams every item across all pages, following `next` links until the
90    /// collection is exhausted. Starts from this page's already-fetched items.
91    pub fn into_stream(self) -> impl Stream<Item = Result<T>> + Send {
92        let fetch = match self.next {
93            Some(url) => Fetch::Follow(url),
94            None => Fetch::Stop,
95        };
96        run_stream(self.client, self.items, fetch)
97    }
98}
99
100/// What the paginating stream should fetch when its current buffer is drained.
101pub(crate) enum Fetch {
102    /// Fetch the first page from a base-relative path + query.
103    First {
104        path: String,
105        query: Vec<(&'static str, String)>,
106    },
107    /// Follow an absolute `next` URL.
108    Follow(String),
109    /// No more pages.
110    Stop,
111}
112
113/// Drives the shared paginating stream: yields buffered items, then fetches the
114/// next page (first request or `next` link) when the buffer empties, stopping
115/// on the first error (yielded once) or when there is no further page.
116pub(crate) fn run_stream<T>(
117    client: Client,
118    initial: Vec<T>,
119    fetch: Fetch,
120) -> impl Stream<Item = Result<T>> + Send
121where
122    T: DeserializeOwned + Send + 'static,
123{
124    struct State<T> {
125        client: Client,
126        buffer: std::vec::IntoIter<T>,
127        fetch: Fetch,
128        done: bool,
129    }
130
131    let state = State {
132        client,
133        buffer: initial.into_iter(),
134        fetch,
135        done: false,
136    };
137
138    futures::stream::unfold(state, |mut st| async move {
139        if st.done {
140            return None;
141        }
142        loop {
143            if let Some(item) = st.buffer.next() {
144                return Some((Ok(item), st));
145            }
146            let resp = match std::mem::replace(&mut st.fetch, Fetch::Stop) {
147                Fetch::Stop => return None,
148                Fetch::First { path, query } => {
149                    st.client
150                        .inner
151                        .send(Method::GET, &path, &query, None, false)
152                        .await
153                }
154                Fetch::Follow(url) => {
155                    st.client
156                        .inner
157                        .send_url(Method::GET, url, &[], None, false)
158                        .await
159                }
160            };
161            match resp.and_then(|r| Page::<T>::from_response(st.client.clone(), &r)) {
162                Ok(page) => {
163                    st.buffer = page.items.into_iter();
164                    st.fetch = match page.next {
165                        Some(url) => Fetch::Follow(url),
166                        None => Fetch::Stop,
167                    };
168                }
169                Err(e) => {
170                    st.done = true;
171                    return Some((Err(e), st));
172                }
173            }
174        }
175    })
176}
177
178/// Parses the `rel="next"` target from the response `Link` headers (RFC 5988).
179///
180/// Handles both one header per relation and a single header carrying several
181/// comma-separated links. Returns the raw (possibly relative) URL.
182pub(crate) fn parse_next_link(headers: &HeaderMap) -> Option<String> {
183    headers
184        .get_all(LINK)
185        .iter()
186        .filter_map(|v| v.to_str().ok())
187        .flat_map(|s| s.split(','))
188        .find_map(|link| {
189            let (url_part, params) = link.split_once(';')?;
190            rel_is_next(params).then(|| {
191                url_part
192                    .trim()
193                    .trim_start_matches('<')
194                    .trim_end_matches('>')
195                    .trim()
196                    .to_owned()
197            })
198        })
199}
200
201/// Whether any `;`-separated parameter is `rel="next"` (case-insensitive, with
202/// or without surrounding quotes).
203fn rel_is_next(params: &str) -> bool {
204    params.split(';').any(|p| match p.split_once('=') {
205        Some((k, v)) => {
206            k.trim().eq_ignore_ascii_case("rel")
207                && v.trim().trim_matches('"').eq_ignore_ascii_case("next")
208        }
209        None => false,
210    })
211}
212
213/// Resolves a `Link` target against the client base: absolute URLs pass through,
214/// base-relative paths are prefixed with `base`.
215fn resolve_url(base: &str, link: &str) -> String {
216    if link.starts_with("http://") || link.starts_with("https://") {
217        link.to_owned()
218    } else {
219        format!("{base}{link}")
220    }
221}
222
223#[cfg(test)]
224mod tests {
225    use super::*;
226    use crate::client::Client;
227    use crate::config::Auth;
228    use crate::dispatch::room_path;
229    use crate::types::Message;
230    use futures::StreamExt;
231    use reqwest::header::{HeaderMap, HeaderValue, LINK};
232    use wiremock::matchers::{method, path, query_param, query_param_is_missing};
233    use wiremock::{Mock, MockServer, ResponseTemplate};
234
235    fn message_json(serial: &str) -> String {
236        format!(
237            r#"{{"serial":"{serial}","version":{{"serial":"{serial}","timestamp":1}},
238               "text":"t","clientId":"a","action":"message.create",
239               "metadata":{{}},"headers":{{}},"timestamp":1}}"#
240        )
241    }
242
243    #[test]
244    fn parses_next_link_and_ignores_other_rels() {
245        let mut headers = HeaderMap::new();
246        headers.append(
247            LINK,
248            HeaderValue::from_static("</chat/v4/rooms/r/messages?cont=first>; rel=\"first\""),
249        );
250        headers.append(
251            LINK,
252            HeaderValue::from_static("</chat/v4/rooms/r/messages?cont=2>; rel=\"next\""),
253        );
254        assert_eq!(
255            parse_next_link(&headers).as_deref(),
256            Some("/chat/v4/rooms/r/messages?cont=2")
257        );
258
259        // A single header carrying several comma-separated links also parses.
260        let mut combined = HeaderMap::new();
261        combined.append(
262            LINK,
263            HeaderValue::from_static("</m?cont=cur>; rel=\"current\", </m?cont=n>; rel=\"next\""),
264        );
265        assert_eq!(parse_next_link(&combined).as_deref(), Some("/m?cont=n"));
266
267        // No next relation → None.
268        let mut only_first = HeaderMap::new();
269        only_first.append(LINK, HeaderValue::from_static("</m?cont=x>; rel=\"first\""));
270        assert_eq!(parse_next_link(&only_first), None);
271    }
272
273    #[test]
274    fn resolves_relative_and_absolute_links() {
275        assert_eq!(
276            resolve_url("https://rest.ably.io", "/chat/v4/rooms/r/messages?cont=2"),
277            "https://rest.ably.io/chat/v4/rooms/r/messages?cont=2"
278        );
279        assert_eq!(
280            resolve_url("https://rest.ably.io", "https://other.host/x?cont=2"),
281            "https://other.host/x?cont=2"
282        );
283    }
284
285    #[tokio::test]
286    async fn manual_next_walks_two_pages_then_none() {
287        let server = MockServer::start().await;
288        // Page 1: no continuation param; emits a relative `next` link.
289        Mock::given(method("GET"))
290            .and(path("/chat/v4/rooms/r/messages"))
291            .and(query_param_is_missing("cont"))
292            .respond_with(
293                ResponseTemplate::new(200)
294                    .append_header("Link", "</chat/v4/rooms/r/messages?cont=2>; rel=\"next\"")
295                    .set_body_string(format!("[{}]", message_json("m1"))),
296            )
297            .mount(&server)
298            .await;
299        // Page 2: continuation param present; no `next` link → last page.
300        Mock::given(method("GET"))
301            .and(path("/chat/v4/rooms/r/messages"))
302            .and(query_param("cont", "2"))
303            .respond_with(
304                ResponseTemplate::new(200).set_body_string(format!("[{}]", message_json("m2"))),
305            )
306            .mount(&server)
307            .await;
308
309        let client = Client::builder(Auth::api_key("k:s"))
310            .host(server.uri())
311            .build();
312        let page1: Page<Message> =
313            Page::fetch_first(client.clone(), room_path("r", "/messages"), Vec::new())
314                .await
315                .unwrap();
316        assert_eq!(page1.items().len(), 1);
317        assert_eq!(page1.items()[0].serial.as_str(), "m1");
318        assert!(page1.has_next());
319
320        let page2 = page1.next().await.unwrap().expect("expected a second page");
321        assert_eq!(page2.items().len(), 1);
322        assert_eq!(page2.items()[0].serial.as_str(), "m2");
323        assert!(!page2.has_next());
324
325        assert!(page2.next().await.unwrap().is_none());
326    }
327
328    #[tokio::test]
329    async fn into_stream_yields_all_items_across_pages() {
330        let server = MockServer::start().await;
331        Mock::given(method("GET"))
332            .and(path("/chat/v4/rooms/r/messages"))
333            .and(query_param_is_missing("cont"))
334            .respond_with(
335                ResponseTemplate::new(200)
336                    .append_header("Link", "</chat/v4/rooms/r/messages?cont=2>; rel=\"next\"")
337                    .set_body_string(format!("[{}]", message_json("m1"))),
338            )
339            .mount(&server)
340            .await;
341        Mock::given(method("GET"))
342            .and(path("/chat/v4/rooms/r/messages"))
343            .and(query_param("cont", "2"))
344            .respond_with(
345                ResponseTemplate::new(200).set_body_string(format!("[{}]", message_json("m2"))),
346            )
347            .mount(&server)
348            .await;
349
350        let client = Client::builder(Auth::api_key("k:s"))
351            .host(server.uri())
352            .build();
353        let page1: Page<Message> =
354            Page::fetch_first(client.clone(), room_path("r", "/messages"), Vec::new())
355                .await
356                .unwrap();
357        let serials: Vec<String> = page1
358            .into_stream()
359            .map(|r| r.unwrap().serial.as_str().to_owned())
360            .collect()
361            .await;
362        assert_eq!(serials, vec!["m1", "m2"]);
363    }
364}
ably_chat/pagination.rs

ably_chat/
pagination.rs
