use serde::Deserialize;
use serde_json::Value;
use crate::error::Result;
use crate::transport::WireRequest;
pub const CURSOR_PARAM: &str = "cursor";
pub const LIMIT_PARAM: &str = "limit";
#[derive(Debug, Clone, Deserialize)]
pub struct Page<T> {
#[serde(default = "Vec::new")]
pub items: Vec<T>,
#[serde(default)]
pub next_cursor: Option<String>,
}
impl<T> Page<T> {
#[must_use]
pub fn next(&self) -> Option<&str> {
self.next_cursor
.as_deref()
.filter(|cursor| !cursor.is_empty())
}
}
impl<T> Default for Page<T> {
fn default() -> Self {
Self {
items: Vec::new(),
next_cursor: None,
}
}
}
#[must_use]
pub fn more_pages(document: &Value) -> bool {
document
.get("next_cursor")
.and_then(Value::as_str)
.is_some_and(|cursor| !cursor.is_empty())
}
pub fn set_cursor(request: &mut WireRequest<'_>, cursor: Option<&str>) {
request.query.retain(|(name, _)| name != CURSOR_PARAM);
if let Some(cursor) = cursor {
request
.query
.push((CURSOR_PARAM.to_owned(), cursor.to_owned()));
}
}
pub async fn walk<T, F, Fut>(mut fetch: F) -> Result<Vec<T>>
where
F: FnMut(Option<String>) -> Fut,
Fut: Future<Output = Result<Page<T>>>,
{
let mut items = Vec::new();
let mut cursor: Option<String> = None;
let mut seen: Vec<String> = Vec::new();
loop {
let mut page = fetch(cursor.clone()).await?;
items.append(&mut page.items);
match page.next() {
Some(next) if !seen.iter().any(|prior| prior == next) => {
seen.push(next.to_owned());
cursor = Some(next.to_owned());
}
_ => return Ok(items),
}
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
use super::*;
use std::cell::RefCell;
fn parse(body: &str) -> Page<i32> {
serde_json::from_str(body).unwrap()
}
#[test]
fn an_empty_cursor_is_the_end_of_the_listing_not_a_fourth_page() {
for raw in [
r#"{"items":[]}"#,
r#"{"items":[],"next_cursor":null}"#,
r#"{"items":[],"next_cursor":""}"#,
] {
let page: Page<Value> = serde_json::from_str(raw).unwrap();
assert_eq!(page.next(), None, "{raw}");
}
let page: Page<Value> = serde_json::from_str(r#"{"items":[],"next_cursor":"c1"}"#).unwrap();
assert_eq!(page.next(), Some("c1"));
}
#[test]
fn a_raw_document_reads_the_same_way_as_a_decoded_page() {
assert!(!more_pages(&serde_json::json!({"items": []})));
assert!(!more_pages(
&serde_json::json!({"items": [], "next_cursor": null})
));
assert!(!more_pages(
&serde_json::json!({"items": [], "next_cursor": ""})
));
assert!(more_pages(
&serde_json::json!({"items": [], "next_cursor": "c1"})
));
}
#[tokio::test]
async fn a_walk_follows_cursors_to_the_end() {
let pages = [
(None, r#"{"items":[1,2],"next_cursor":"c1"}"#),
(Some("c1"), r#"{"items":[3],"next_cursor":"c2"}"#),
(Some("c2"), r#"{"items":[4],"next_cursor":null}"#),
];
let asked: RefCell<Vec<Option<String>>> = RefCell::new(Vec::new());
let fetch = |cursor: Option<String>| {
asked.borrow_mut().push(cursor.clone());
let page = pages
.iter()
.find(|(want, _)| want.map(ToOwned::to_owned) == cursor)
.map(|(_, body)| parse(body))
.unwrap();
std::future::ready(Ok(page))
};
let items: Vec<i32> = walk(fetch).await.unwrap();
assert_eq!(items, vec![1, 2, 3, 4]);
assert_eq!(
asked.into_inner(),
vec![None, Some("c1".to_owned()), Some("c2".to_owned())],
);
}
#[tokio::test]
async fn a_repeated_cursor_stops_the_walk_rather_than_hanging() {
let calls = RefCell::new(0_usize);
let fetch = |_| {
*calls.borrow_mut() += 1;
std::future::ready(Ok(parse(r#"{"items":[1],"next_cursor":"stuck"}"#)))
};
let items: Vec<i32> = walk(fetch).await.unwrap();
assert_eq!(*calls.borrow(), 2, "the repeat must end the walk");
assert_eq!(items, vec![1, 1]);
}
#[test]
fn setting_a_cursor_replaces_rather_than_appends() {
let mut request = WireRequest {
method: "GET",
path: "/v1/sessions",
query: vec![
("limit".to_owned(), "25".to_owned()),
(CURSOR_PARAM.to_owned(), "old".to_owned()),
],
..Default::default()
};
set_cursor(&mut request, Some("new"));
assert_eq!(
request.query,
vec![
("limit".to_owned(), "25".to_owned()),
(CURSOR_PARAM.to_owned(), "new".to_owned()),
],
);
set_cursor(&mut request, None);
assert_eq!(request.query, vec![("limit".to_owned(), "25".to_owned())]);
}
}