berkus
/
actix-net
mirror of https://github.com/fafhrd91/actix-net
1
0
Fork 0
Code Issues Releases Wiki Activity

remove timeout mod

This commit is contained in:
Rob Ede 2021-03-30 03:09:35 +01:00
parent fe01601ff3
commit 7a760f9063
No known key found for this signature in database
GPG Key ID: 97C636207D3EF933
4 changed files with 2 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
actix-utils/CHANGES.md
View File

@ -3,6 +3,7 @@
## Unreleased - 2021-xx-xx
* Moved `mpsc` to own crate `local-channel`. [#301]
* Moved `task::LocalWaker` to own crate `local-waker`. [#301]
* Remove `timeout` module. [#301]
* Expose `future` mod with `ready` and `poll_fn` helpers. [#301]
* `SendError` inner field is now public. [#286]
* Rename `Dispatcher::{get_sink => tx}`. [#286]

6
actix-utils/Cargo.toml
View File

@ -16,14 +16,8 @@ name = "actix_utils"
path = "src/lib.rs"
[dependencies]
actix-rt = { version = "2.0.0", default-features = false }
actix-service = "2.0.0-beta.5"
local-waker = "0.1"
log = "0.4"
pin-project-lite = "0.2.0"
[dev-dependencies]
actix-rt = "2.0.0"
futures-core = { version = "0.3.7", default-features = false, features = ["alloc"] }
futures-util = { version = "0.3.7", default-features = false }

2
actix-utils/src/lib.rs
View File

@ -1,5 +1,6 @@
//! Various network related services and utilities for the Actix ecosystem.
#![no_std]
#![deny(rust_2018_idioms, nonstandard_style)]
#![allow(clippy::type_complexity)]
#![doc(html_logo_url = "https://actix.rs/img/logo.png")]
@ -9,4 +10,3 @@ extern crate alloc;
pub mod counter;
pub mod future;
pub mod timeout;

259
actix-utils/src/timeout.rs
View File

@ -1,259 +0,0 @@
//! Service that applies a timeout to requests.
//!
//! If the response does not complete within the specified timeout, the response will be aborted.
use core::{
fmt,
future::Future,
marker::PhantomData,
pin::Pin,
task::{Context, Poll},
time,
};
use actix_rt::time::{sleep, Sleep};
use actix_service::{IntoService, Service, Transform};
use pin_project_lite::pin_project;
/// Applies a timeout to requests.
#[derive(Debug)]
pub struct Timeout<E = ()> {
timeout: time::Duration,
_t: PhantomData<E>,
}
/// Service or timeout error.
pub enum TimeoutError<E> {
/// Inner service error.
Service(E),
/// Timeout during service call.
Timeout,
}
impl<E> From<E> for TimeoutError<E> {
fn from(err: E) -> Self {
TimeoutError::Service(err)
}
}
impl<E: fmt::Debug> fmt::Debug for TimeoutError<E> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
TimeoutError::Service(e) => write!(f, "TimeoutError::Service({:?})", e),
TimeoutError::Timeout => write!(f, "TimeoutError::Timeout"),
}
}
}
impl<E: fmt::Display> fmt::Display for TimeoutError<E> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
TimeoutError::Service(err) => err.fmt(f),
TimeoutError::Timeout => write!(f, "timeout during service call"),
}
}
}
impl<E: PartialEq> PartialEq for TimeoutError<E> {
fn eq(&self, other: &TimeoutError<E>) -> bool {
match self {
TimeoutError::Service(e1) => match other {
TimeoutError::Service(e2) => e1 == e2,
TimeoutError::Timeout => false,
},
TimeoutError::Timeout => matches!(other, TimeoutError::Timeout),
}
}
}
impl<E> Timeout<E> {
pub fn new(timeout: time::Duration) -> Self {
Timeout {
timeout,
_t: PhantomData,
}
}
}
impl<E> Clone for Timeout<E> {
fn clone(&self) -> Self {
Timeout::new(self.timeout)
}
}
impl<S, E, Req> Transform<S, Req> for Timeout<E>
where
S: Service<Req>,
{
type Response = S::Response;
type Error = TimeoutError<S::Error>;
type Transform = TimeoutService<S, Req>;
type InitError = E;
type Future = TimeoutFuture<Self::Transform, Self::InitError>;
fn new_transform(&self, service: S) -> Self::Future {
let service = TimeoutService {
service,
timeout: self.timeout,
_phantom: PhantomData,
};
TimeoutFuture {
service: Some(service),
_err: PhantomData,
}
}
}
pub struct TimeoutFuture<T, E> {
service: Option<T>,
_err: PhantomData<E>,
}
impl<T, E> Unpin for TimeoutFuture<T, E> {}
impl<T, E> Future for TimeoutFuture<T, E> {
type Output = Result<T, E>;
fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Self::Output> {
Poll::Ready(Ok(self.get_mut().service.take().unwrap()))
}
}
/// Applies a timeout to requests.
#[derive(Debug, Clone)]
pub struct TimeoutService<S, Req> {
service: S,
timeout: time::Duration,
_phantom: PhantomData<Req>,
}
impl<S, Req> TimeoutService<S, Req>
where
S: Service<Req>,
{
pub fn new<U>(timeout: time::Duration, service: U) -> Self
where
U: IntoService<S, Req>,
{
TimeoutService {
timeout,
service: service.into_service(),
_phantom: PhantomData,
}
}
}
impl<S, Req> Service<Req> for TimeoutService<S, Req>
where
S: Service<Req>,
{
type Response = S::Response;
type Error = TimeoutError<S::Error>;
type Future = TimeoutServiceResponse<S, Req>;
actix_service::forward_ready!(service);
fn call(&self, request: Req) -> Self::Future {
TimeoutServiceResponse {
fut: self.service.call(request),
sleep: sleep(self.timeout),
}
}
}
pin_project! {
/// `TimeoutService` response future.
#[derive(Debug)]
pub struct TimeoutServiceResponse<S, Req>
where
S: Service<Req>
{
#[pin]
fut: S::Future,
#[pin]
sleep: Sleep,
}
}
impl<S, Req> Future for TimeoutServiceResponse<S, Req>
where
S: Service<Req>,
{
type Output = Result<S::Response, TimeoutError<S::Error>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
// first try polling the future
if let Poll::Ready(res) = this.fut.poll(cx) {
return match res {
Ok(v) => Poll::Ready(Ok(v)),
Err(e) => Poll::Ready(Err(TimeoutError::Service(e))),
};
}
// now check the sleep
this.sleep.poll(cx).map(|_| Err(TimeoutError::Timeout))
}
}
#[cfg(test)]
mod tests {
use core::time::Duration;
use super::*;
use actix_service::{apply, fn_factory, Service, ServiceFactory};
use futures_core::future::LocalBoxFuture;
struct SleepService(Duration);
impl Service<()> for SleepService {
type Response = ();
type Error = ();
type Future = LocalBoxFuture<'static, Result<(), ()>>;
actix_service::always_ready!();
fn call(&self, _: ()) -> Self::Future {
let sleep = actix_rt::time::sleep(self.0);
Box::pin(async move {
sleep.await;
Ok(())
})
}
}
#[actix_rt::test]
async fn test_success() {
let resolution = Duration::from_millis(100);
let wait_time = Duration::from_millis(50);
let timeout = TimeoutService::new(resolution, SleepService(wait_time));
assert_eq!(timeout.call(()).await, Ok(()));
}
#[actix_rt::test]
async fn test_timeout() {
let resolution = Duration::from_millis(100);
let wait_time = Duration::from_millis(500);
let timeout = TimeoutService::new(resolution, SleepService(wait_time));
assert_eq!(timeout.call(()).await, Err(TimeoutError::Timeout));
}
#[actix_rt::test]
async fn test_timeout_new_service() {
let resolution = Duration::from_millis(100);
let wait_time = Duration::from_millis(500);
let timeout = apply(
Timeout::new(resolution),
fn_factory(|| async { Ok::<_, ()>(SleepService(wait_time)) }),
);
let srv = timeout.new_service(&()).await.unwrap();
assert_eq!(srv.call(()).await, Err(TimeoutError::Timeout));
}
}