2020-06-12 11:42:28 -07:00
|
|
|
//! Future types for the `Batch` middleware.
|
2020-06-12 10:22:08 -07:00
|
|
|
|
|
|
|
use super::{error::Closed, message};
|
|
|
|
use futures_core::ready;
|
2020-06-16 13:56:18 -07:00
|
|
|
use pin_project::pin_project;
|
2020-06-12 10:22:08 -07:00
|
|
|
use std::{
|
2020-06-17 14:39:10 -07:00
|
|
|
fmt::Debug,
|
2020-06-12 10:22:08 -07:00
|
|
|
future::Future,
|
|
|
|
pin::Pin,
|
|
|
|
task::{Context, Poll},
|
|
|
|
};
|
2020-06-17 14:39:10 -07:00
|
|
|
use tower::Service;
|
2020-06-12 10:22:08 -07:00
|
|
|
|
2020-06-12 11:42:28 -07:00
|
|
|
/// Future that completes when the batch processing is complete.
|
2020-06-12 10:22:08 -07:00
|
|
|
#[pin_project]
|
|
|
|
#[derive(Debug)]
|
2020-06-17 14:39:10 -07:00
|
|
|
pub struct ResponseFuture<T, E, R>
|
|
|
|
where
|
|
|
|
T: Service<crate::BatchControl<R>>,
|
|
|
|
{
|
2020-06-12 10:22:08 -07:00
|
|
|
#[pin]
|
2020-06-17 14:39:10 -07:00
|
|
|
state: ResponseState<T, E, R>,
|
2020-06-12 10:22:08 -07:00
|
|
|
}
|
|
|
|
|
2020-06-16 13:56:18 -07:00
|
|
|
#[pin_project(project = ResponseStateProj)]
|
2020-06-17 14:39:10 -07:00
|
|
|
enum ResponseState<T, E, R>
|
|
|
|
where
|
|
|
|
T: Service<crate::BatchControl<R>>,
|
|
|
|
{
|
|
|
|
Failed(Option<E>),
|
|
|
|
Rx(#[pin] message::Rx<T::Future, T::Error>),
|
|
|
|
Poll(#[pin] T::Future),
|
|
|
|
}
|
|
|
|
|
|
|
|
impl<T, E, R> Debug for ResponseState<T, E, R>
|
|
|
|
where
|
|
|
|
T: Service<crate::BatchControl<R>>,
|
|
|
|
{
|
|
|
|
fn fmt(&self, _f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
|
|
todo!()
|
|
|
|
}
|
2020-06-12 10:22:08 -07:00
|
|
|
}
|
|
|
|
|
2020-06-17 14:39:10 -07:00
|
|
|
impl<T, E, R> ResponseFuture<T, E, R>
|
|
|
|
where
|
|
|
|
T: Service<crate::BatchControl<R>>,
|
|
|
|
{
|
|
|
|
pub(crate) fn new(rx: message::Rx<T::Future, T::Error>) -> Self {
|
2020-06-12 10:22:08 -07:00
|
|
|
ResponseFuture {
|
|
|
|
state: ResponseState::Rx(rx),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-06-17 14:39:10 -07:00
|
|
|
pub(crate) fn failed(err: E) -> Self {
|
2020-06-12 10:22:08 -07:00
|
|
|
ResponseFuture {
|
|
|
|
state: ResponseState::Failed(Some(err)),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-06-17 14:39:10 -07:00
|
|
|
impl<S, E2, R> Future for ResponseFuture<S, E2, R>
|
2020-06-12 10:22:08 -07:00
|
|
|
where
|
2020-06-17 14:39:10 -07:00
|
|
|
S: Service<crate::BatchControl<R>>,
|
|
|
|
S::Future: Future<Output = Result<S::Response, S::Error>>,
|
|
|
|
S::Error: Into<E2>,
|
|
|
|
crate::error::Closed: Into<E2>,
|
2020-06-12 10:22:08 -07:00
|
|
|
{
|
2020-06-17 14:39:10 -07:00
|
|
|
type Output = Result<S::Response, E2>;
|
2020-06-12 10:22:08 -07:00
|
|
|
|
|
|
|
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
|
|
|
let mut this = self.project();
|
|
|
|
|
|
|
|
loop {
|
|
|
|
match this.state.as_mut().project() {
|
2020-06-16 13:56:18 -07:00
|
|
|
ResponseStateProj::Failed(e) => {
|
2020-06-12 10:22:08 -07:00
|
|
|
return Poll::Ready(Err(e.take().expect("polled after error")));
|
|
|
|
}
|
2020-06-16 13:56:18 -07:00
|
|
|
ResponseStateProj::Rx(rx) => match ready!(rx.poll(cx)) {
|
2020-06-12 10:22:08 -07:00
|
|
|
Ok(Ok(f)) => this.state.set(ResponseState::Poll(f)),
|
|
|
|
Ok(Err(e)) => return Poll::Ready(Err(e.into())),
|
|
|
|
Err(_) => return Poll::Ready(Err(Closed::new().into())),
|
|
|
|
},
|
2020-06-16 13:56:18 -07:00
|
|
|
ResponseStateProj::Poll(fut) => return fut.poll(cx).map_err(Into::into),
|
2020-06-12 10:22:08 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|