use crate::auth::AuthToken;
use crate::common::ContinuationParams;
use crate::parse::ParseFrom;
use crate::query::{GetContinuationsQuery, PostMethod, PostQuery, Query, QueryMethod};
use crate::{ProcessedResult, Result};
use futures::Stream;
use std::fmt::Debug;
pub trait Continuable<Q>: Sized {
fn take_continuation_params(&mut self) -> Option<ContinuationParams<'static>>;
fn parse_continuation(p: ProcessedResult<GetContinuationsQuery<'_, Q>>) -> Result<Self>;
}
impl<'a, T, Q> ParseFrom<GetContinuationsQuery<'a, Q>> for T
where
T: Continuable<Q>,
T: Debug,
{
fn parse_from(p: ProcessedResult<GetContinuationsQuery<'a, Q>>) -> Result<Self> {
T::parse_continuation(p)
}
}
pub(crate) fn stream<'a, Q, A>(
query: &'a Q,
client: &'a crate::client::Client,
tok: &'a A,
) -> impl Stream<Item = Result<Q::Output>> + 'a
where
A: AuthToken,
Q: Query<A>,
Q: PostQuery,
Q::Output: Continuable<Q>,
{
futures::stream::unfold(
(false, None::<GetContinuationsQuery<Q>>),
move |(first, maybe_next_query)| async move {
if !first {
let first_res: Result<Q::Output> = Q::Method::call(query, client, tok)
.await
.and_then(|res| res.process())
.and_then(|res| res.parse_into());
match first_res {
Ok(mut first) => {
let maybe_next_query = GetContinuationsQuery::<Q>::new(&mut first, query);
return Some((Ok(first), (true, maybe_next_query)));
}
Err(e) => return Some((Err(e), (true, None))),
}
}
if let Some(next_query) = maybe_next_query {
let next = PostMethod::call(&next_query, client, tok)
.await
.and_then(|res| res.process())
.and_then(|res| res.parse_into());
match next {
Ok(mut next) => {
let maybe_next_query = GetContinuationsQuery::<Q>::new(&mut next, query);
return Some((Ok(next), (true, maybe_next_query)));
}
Err(e) => return Some((Err(e), (true, None))),
}
}
None
},
)
}