route dispatch with timeout

legacy
Avril 4 years ago
parent ee3cdbf8bb
commit d06dee54eb
Signed by: flanchan
GPG Key ID: 284488987C31F630

@ -20,7 +20,7 @@ impl<T> OpaqueDebug<T>
}
/// Consume into the value
#[inline] pub const fn into_inner(self) -> T
#[inline] pub fn into_inner(self) -> T
{
self.0
}

@ -10,10 +10,12 @@ use std::{
};
use tokio::{
sync::{
mpsc,
broadcast,
notify,
mpsc::{
self,
error::SendTimeoutError,
},
},
time,
};
use futures::{
future::{
@ -96,20 +98,9 @@ impl Router
///
/// # Returns
/// When one or more dispatchers match but faile, `Err` is returned. Inside the `Err` tuple is the amount of successful dispatches, and also a vector containing the indecies of the failed hook sends.
pub async fn dispatch(&mut self, method: &Method, uri: impl AsRef<str>, timeout: impl Future) -> Result<usize, (usize, Vec<Index>)>
pub async fn dispatch(&mut self, method: &Method, uri: impl AsRef<str>, timeout: Option<time::Duration>) -> Result<usize, (usize, Vec<Index>)>
{
let string = uri.as_ref();
let (tcx, trx) = broadcast::channel(1);
tokio::pin!(timeout);
let begin_to = tokio::sync::Barrier::new(self.routes.len()+1);
let timeout = async {
timeout.await;
begin_to.wait().await;
tcx.send(());
};
let mut timeouts = iter::once(trx)
.chain(iter::repeat_with(|| tcx.subscribe()));
let output = async {
let mut success=0usize;
let vec: Vec<_> =
future::join_all(self.routes.iter_mut()
@ -119,25 +110,27 @@ impl Router
_ => {
if route.is_match(string) {
trace!("{:?} @{}: -> {}",i, route.as_string(), string);
let mut timeout = timeouts.next().unwrap();
Some(async move {
match tokio::select!{
_ = timeout.recv() => {
None
let timeout = timeout.clone();
macro_rules! send {
() => {
match timeout {
None => sender.send(string.to_owned()).await
.map_err(|e| SendTimeoutError::Closed(e.0)),
Some(time) => sender.send_timeout(string.to_owned(), time).await
}
s = sender.send(string.to_owned()) => {
Some(s)
}
} {
Some(Err(er)) => {
warn!("{:?}: Dispatch failed on hooked route for {}", i, er.0);
};
Some(async move {
match send!() {
Err(SendTimeoutError::Closed(er)) => {
error!("{:?}: Dispatch failed on hooked route for {}", i, er);
Err(i)
},
Some(_) => Ok(()),
None => {
warn!("{:?}: Dispatch timed out on hooked route", i);
Err(SendTimeoutError::Timeout(er)) => {
warn!("{:?}: Dispatch timed out on hooked route for {}", i, er);
Err(i)
},
_ => Ok(()),
}
})
} else {
@ -152,10 +145,6 @@ impl Router
}
res.err()
}).collect();
(success, vec)
};
tokio::pin!(output);
let (_, (success, vec)) = future::join(timeout, output).await;
if vec.len() > 0 {
Err((success, vec))
} else {

Loading…
Cancel
Save