use crate::{ResponseMetadata, Status, future::cancellable};
use base_scheduler::Cancellation;
use std::{error::Error, fmt, future::Future};
pub trait InterimHandler: Send {
type Error: Error + Send + Sync + 'static;
fn on_interim_response(
&mut self,
response: ResponseMetadata,
cancellation: &Cancellation,
) -> impl Future<Output = Result<(), Self::Error>> + Send;
}
impl<F, Fut, E> InterimHandler for F
where
F: FnMut(ResponseMetadata) -> Fut + Send,
Fut: Future<Output = Result<(), E>> + Send,
E: Error + Send + Sync + 'static,
{
type Error = E;
async fn on_interim_response(
&mut self,
response: ResponseMetadata,
_: &Cancellation,
) -> Result<(), E> {
self(response).await
}
}
#[derive(Debug)]
pub enum InterimError<E> {
InvalidStatus(Status),
Finished,
Cancelled,
Handler(E),
}
impl<E: fmt::Display> fmt::Display for InterimError<E> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidStatus(status) => write!(formatter, "{status} is not an interim response"),
Self::Finished => formatter.write_str("interim responses have finished"),
Self::Cancelled => formatter.write_str("interim response cancelled"),
Self::Handler(error) => error.fmt(formatter),
}
}
}
impl<E: Error + 'static> Error for InterimError<E> {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Handler(error) => error.source(),
_ => None,
}
}
}
pub struct InterimResponses<H> {
handler: H,
finished: bool,
}
impl<H> InterimResponses<H> {
pub fn new(handler: H) -> Self {
Self {
handler,
finished: false,
}
}
pub fn finish(&mut self) {
self.finished = true;
}
pub fn is_finished(&self) -> bool {
self.finished
}
pub fn into_inner(self) -> H {
self.handler
}
}
impl<H: InterimHandler> InterimResponses<H> {
pub async fn send_interim_response(
&mut self,
response: ResponseMetadata,
cancellation: &Cancellation,
) -> Result<(), InterimError<H::Error>> {
cancellation.check().map_err(|_| InterimError::Cancelled)?;
if self.finished {
return Err(InterimError::Finished);
}
if !response.status.is_informational() || response.status.is_final() {
return Err(InterimError::InvalidStatus(response.status));
}
self.finished = true;
cancellable(
self.handler.on_interim_response(response, cancellation),
cancellation,
)
.await
.map_err(|_| InterimError::Cancelled)?
.map_err(InterimError::Handler)?;
self.finished = false;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io;
#[test]
fn handler_errors_are_transparent() {
let error = InterimError::Handler(crate::body::BodyError::new(io::Error::other(
"observer failed",
)));
assert_eq!(error.to_string(), "body failed");
assert_eq!(error.source().unwrap().to_string(), "observer failed");
}
}