actix/actix-net
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-net synced 2024-11-23 20:51:06 +01:00
Code Issues Releases Wiki Activity

refactor Resolver service

Browse Source
This commit is contained in:
Nikolay Kim 2018-10-29 20:29:47 -07:00
parent 67961f8a36
commit dc19a9f862
14 changed files with 107 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
examples/basic.rs
View File

@ -80,8 +80,7 @@ fn main() {
future::ok(())
})
},
)
.unwrap()
).unwrap()
.start();
sys.run();

6
examples/ssl.rs
View File

@ -24,7 +24,8 @@ struct ServiceState {
}
fn service<T: AsyncRead + AsyncWrite>(
st: &mut ServiceState, _: T,
st: &mut ServiceState,
_: T,
) -> impl Future<Item = (), Error = ()> {
let num = st.num.fetch_add(1, Ordering::Relaxed);
println!("got ssl connection {:?}", num);
@ -60,8 +61,7 @@ fn main() {
println!("got ssl connection {:?}", num);
future::ok(())
})
})
.unwrap()
}).unwrap()
.start();
sys.run();

3
src/codec/framed.rs
View File

@ -176,7 +176,8 @@ where
type SinkError = U::Error;
fn start_send(
&mut self, item: Self::SinkItem,
&mut self,
item: Self::SinkItem,
) -> StartSend<Self::SinkItem, Self::SinkError> {
self.inner.get_mut().start_send(item)
}

3
src/codec/framed2.rs
View File

@ -191,7 +191,8 @@ where
type SinkError = E::Error;
fn start_send(
&mut self, item: Self::SinkItem,
&mut self,
item: Self::SinkItem,
) -> StartSend<Self::SinkItem, Self::SinkError> {
self.inner.get_mut().start_send(item)
}

3
src/codec/framed_read.rs
View File

@ -98,7 +98,8 @@ where
type SinkError = T::SinkError;
fn start_send(
&mut self, item: Self::SinkItem,
&mut self,
item: Self::SinkItem,
) -> StartSend<Self::SinkItem, Self::SinkError> {
self.inner.inner.0.start_send(item)
}

3
src/codec/framed_write.rs
View File

@ -197,8 +197,7 @@ where
io::ErrorKind::WriteZero,
"failed to \
write frame to transport",
)
.into());
).into());
}
// TODO: Add a way to `bytes` to do this w/o returning the drained

67
src/connector.rs
View File

@ -11,7 +11,7 @@ use tokio_tcp::{ConnectFuture, TcpStream};
use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
use trust_dns_resolver::system_conf::read_system_conf;
use super::resolver::{HostAware, Resolver, ResolverError, ResolverFuture};
use super::resolver::{HostAware, ResolveError, Resolver, ResolverFuture};
use super::service::{NewService, Service};
// #[derive(Fail, Debug)]
@ -19,9 +19,9 @@ use super::service::{NewService, Service};
pub enum ConnectorError {
/// Failed to resolve the hostname
// #[fail(display = "Failed resolving hostname: {}", _0)]
Resolver(ResolverError),
Resolver(ResolveError),
/// Not dns records
/// No dns records
// #[fail(display = "Invalid input: {}", _0)]
NoRecords,
@ -29,17 +29,21 @@ pub enum ConnectorError {
// #[fail(display = "Timeout out while establishing connection")]
Timeout,
/// Invalid input
InvalidInput,
/// Connection io error
// #[fail(display = "{}", _0)]
IoError(io::Error),
}
impl From<ResolverError> for ConnectorError {
fn from(err: ResolverError) -> Self {
impl From<ResolveError> for ConnectorError {
fn from(err: ResolveError) -> Self {
ConnectorError::Resolver(err)
}
}
/// Connect request
#[derive(Eq, PartialEq, Debug, Hash)]
pub struct Connect {
pub host: String,
@ -48,6 +52,7 @@ pub struct Connect {
}
impl Connect {
/// Create new `Connect` instance.
pub fn new<T: AsRef<str>>(host: T, port: u16) -> Connect {
Connect {
port,
@ -56,20 +61,19 @@ impl Connect {
}
}
/// split the string by ':' and convert the second part to u16
pub fn parse<T: AsRef<str>>(host: T) -> Option<Connect> {
/// Create `Connect` instance by spliting the string by ':' and convert the second part to u16
pub fn with<T: AsRef<str>>(host: T) -> Result<Connect, ConnectorError> {
let mut parts_iter = host.as_ref().splitn(2, ':');
if let Some(host) = parts_iter.next() {
let port_str = parts_iter.next().unwrap_or("");
if let Ok(port) = port_str.parse::<u16>() {
return Some(Connect {
port,
host: host.to_owned(),
timeout: Duration::from_secs(1),
});
}
}
None
let host = parts_iter.next().ok_or(ConnectorError::InvalidInput)?;
let port_str = parts_iter.next().unwrap_or("");
let port = port_str
.parse::<u16>()
.map_err(|_| ConnectorError::InvalidInput)?;
Ok(Connect {
port,
host: host.to_owned(),
timeout: Duration::from_secs(1),
})
}
/// Set connect timeout
@ -93,6 +97,7 @@ impl fmt::Display for Connect {
}
}
/// Tcp connector
pub struct Connector {
resolver: Resolver<Connect>,
}
@ -110,12 +115,14 @@ impl Default for Connector {
}
impl Connector {
/// Create new connector with resolver configuration
pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
Connector {
resolver: Resolver::new(cfg, opts),
}
}
/// Create new connector with custom resolver
pub fn with_resolver(
resolver: Resolver<Connect>,
) -> impl Service<Request = Connect, Response = (Connect, TcpStream), Error = ConnectorError>
@ -123,17 +130,10 @@ impl Connector {
Connector { resolver }
}
pub fn new_service<E>() -> impl NewService<
Request = Connect,
Response = (Connect, TcpStream),
Error = ConnectorError,
InitError = E,
> + Clone {
|| -> FutureResult<Connector, E> { ok(Connector::default()) }
}
/// Create new default connector service
pub fn new_service_with_config<E>(
cfg: ResolverConfig, opts: ResolverOpts,
cfg: ResolverConfig,
opts: ResolverOpts,
) -> impl NewService<
Request = Connect,
Response = (Connect, TcpStream),
@ -185,19 +185,14 @@ impl Future for ConnectorFuture {
return fut.poll();
}
match self.fut.poll().map_err(ConnectorError::from)? {
Async::Ready((req, _, mut addrs)) => {
Async::Ready((req, mut addrs)) => {
if addrs.is_empty() {
Err(ConnectorError::NoRecords)
} else {
for addr in &mut addrs {
match addr {
SocketAddr::V4(ref mut addr) if addr.port() == 0 => {
addr.set_port(req.port)
}
SocketAddr::V6(ref mut addr) if addr.port() == 0 => {
addr.set_port(req.port)
}
_ => (),
SocketAddr::V4(ref mut addr) => addr.set_port(req.port),
SocketAddr::V6(ref mut addr) => addr.set_port(req.port),
}
}
self.fut2 = Some(TcpConnector::new(req, addrs));

5
src/lib.rs
View File

@ -9,7 +9,10 @@
#![cfg_attr(
feature = "cargo-clippy",
allow(declare_interior_mutable_const, borrow_interior_mutable_const)
allow(
declare_interior_mutable_const,
borrow_interior_mutable_const
)
)]
#[macro_use]

82
src/resolver.rs
View File

@ -6,19 +6,13 @@ use futures::{Async, Future, Poll};
use tokio_current_thread::spawn;
use trust_dns_resolver::config::{ResolverConfig, ResolverOpts};
use trust_dns_resolver::error::ResolveError;
pub use trust_dns_resolver::error::ResolveError;
use trust_dns_resolver::lookup_ip::LookupIpFuture;
use trust_dns_resolver::system_conf::read_system_conf;
use trust_dns_resolver::{AsyncResolver, Background};
use super::service::Service;
#[derive(Debug)]
pub enum ResolverError {
Resolve(ResolveError),
InvalidInput,
}
pub trait HostAware {
fn host(&self) -> &str;
}
@ -75,8 +69,8 @@ impl<T> Clone for Resolver<T> {
impl<T: HostAware> Service for Resolver<T> {
type Request = T;
type Response = (T, String, VecDeque<SocketAddr>);
type Error = ResolverError;
type Response = (T, VecDeque<SocketAddr>);
type Error = ResolveError;
type Future = ResolverFuture<T>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
@ -84,7 +78,7 @@ impl<T: HostAware> Service for Resolver<T> {
}
fn call(&mut self, req: Self::Request) -> Self::Future {
ResolverFuture::new(req, 0, &self.resolver)
ResolverFuture::new(req, &self.resolver)
}
}
@ -92,78 +86,38 @@ impl<T: HostAware> Service for Resolver<T> {
/// Resolver future
pub struct ResolverFuture<T> {
req: Option<T>,
port: u16,
lookup: Option<Background<LookupIpFuture>>,
addrs: Option<VecDeque<SocketAddr>>,
error: Option<ResolverError>,
host: Option<String>,
}
impl<T: HostAware> ResolverFuture<T> {
pub fn new(addr: T, port: u16, resolver: &AsyncResolver) -> Self {
pub fn new(addr: T, resolver: &AsyncResolver) -> Self {
// we need to do dns resolution
match ResolverFuture::<T>::parse(addr.host(), port) {
Ok((host, port)) => {
let lookup = Some(resolver.lookup_ip(host.as_str()));
ResolverFuture {
port,
lookup,
req: Some(addr),
host: Some(host.to_owned()),
addrs: None,
error: None,
}
}
Err(err) => ResolverFuture {
port,
req: None,
host: None,
lookup: None,
addrs: None,
error: Some(err),
},
let lookup = Some(resolver.lookup_ip(addr.host()));
ResolverFuture {
lookup,
req: Some(addr),
addrs: None,
}
}
fn parse(addr: &str, port: u16) -> Result<(String, u16), ResolverError> {
// split the string by ':' and convert the second part to u16
let mut parts_iter = addr.splitn(2, ':');
let host = parts_iter.next().ok_or(ResolverError::InvalidInput)?;
let port_str = parts_iter.next().unwrap_or("");
let port: u16 = port_str.parse().unwrap_or(port);
Ok((host.to_owned(), port))
}
}
impl<T: HostAware> Future for ResolverFuture<T> {
type Item = (T, String, VecDeque<SocketAddr>);
type Error = ResolverError;
type Item = (T, VecDeque<SocketAddr>);
type Error = ResolveError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
if let Some(err) = self.error.take() {
Err(err)
} else if let Some(addrs) = self.addrs.take() {
Ok(Async::Ready((
self.req.take().unwrap(),
self.host.take().unwrap(),
addrs,
)))
if let Some(addrs) = self.addrs.take() {
Ok(Async::Ready((self.req.take().unwrap(), addrs)))
} else {
match self.lookup.as_mut().unwrap().poll() {
Ok(Async::NotReady) => Ok(Async::NotReady),
Ok(Async::Ready(ips)) => {
let addrs: VecDeque<_> = ips
.iter()
.map(|ip| SocketAddr::new(ip, self.port))
.collect();
Ok(Async::Ready((
self.req.take().unwrap(),
self.host.take().unwrap(),
addrs,
)))
let addrs: VecDeque<_> =
ips.iter().map(|ip| SocketAddr::new(ip, 0)).collect();
Ok(Async::Ready((self.req.take().unwrap(), addrs)))
}
Err(err) => Err(ResolverError::Resolve(err)),
Err(err) => Err(err),
}
}
}

19
src/server/accept.rs
View File

@ -87,7 +87,9 @@ impl AcceptLoop {
}
pub(crate) fn start(
&mut self, socks: Vec<(Token, net::TcpListener)>, workers: Vec<WorkerClient>,
&mut self,
socks: Vec<(Token, net::TcpListener)>,
workers: Vec<WorkerClient>,
) -> mpsc::UnboundedReceiver<ServerCommand> {
let (tx, rx) = self.srv.take().expect("Can not re-use AcceptInfo");
@ -135,9 +137,12 @@ fn connection_error(e: &io::Error) -> bool {
impl Accept {
#![cfg_attr(feature = "cargo-clippy", allow(too_many_arguments))]
pub(crate) fn start(
rx: sync_mpsc::Receiver<Command>, cmd_reg: mio::Registration,
notify_reg: mio::Registration, socks: Vec<(Token, net::TcpListener)>,
srv: mpsc::UnboundedSender<ServerCommand>, workers: Vec<WorkerClient>,
rx: sync_mpsc::Receiver<Command>,
cmd_reg: mio::Registration,
notify_reg: mio::Registration,
socks: Vec<(Token, net::TcpListener)>,
srv: mpsc::UnboundedSender<ServerCommand>,
workers: Vec<WorkerClient>,
) {
let sys = System::current();
@ -173,8 +178,10 @@ impl Accept {
}
fn new(
rx: sync_mpsc::Receiver<Command>, socks: Vec<(Token, net::TcpListener)>,
workers: Vec<WorkerClient>, srv: mpsc::UnboundedSender<ServerCommand>,
rx: sync_mpsc::Receiver<Command>,
socks: Vec<(Token, net::TcpListener)>,
workers: Vec<WorkerClient>,
srv: mpsc::UnboundedSender<ServerCommand>,
) -> Accept {
// Create a poll instance
let poll = match mio::Poll::new() {

10
src/server/server.rs
View File

@ -137,7 +137,10 @@ impl Server {
/// Add new service to server
pub fn listen<F, N: AsRef<str>>(
mut self, name: N, lst: net::TcpListener, factory: F,
mut self,
name: N,
lst: net::TcpListener,
factory: F,
) -> Self
where
F: StreamServiceFactory,
@ -151,7 +154,10 @@ impl Server {
/// Add new service to server
pub fn listen2<F, N: AsRef<str>>(
mut self, name: N, lst: net::TcpListener, factory: F,
mut self,
name: N,
lst: net::TcpListener,
factory: F,
) -> Self
where
F: ServiceFactory,

13
src/server/worker.rs
View File

@ -61,7 +61,9 @@ pub(crate) struct WorkerClient {
impl WorkerClient {
pub fn new(
idx: usize, tx: UnboundedSender<WorkerCommand>, avail: WorkerAvailability,
idx: usize,
tx: UnboundedSender<WorkerCommand>,
avail: WorkerAvailability,
) -> Self {
WorkerClient { idx, tx, avail }
}
@ -128,8 +130,10 @@ pub(crate) struct Worker {
impl Worker {
pub(crate) fn start(
rx: UnboundedReceiver<WorkerCommand>, factories: Vec<Box<InternalServiceFactory>>,
availability: WorkerAvailability, shutdown_timeout: time::Duration,
rx: UnboundedReceiver<WorkerCommand>,
factories: Vec<Box<InternalServiceFactory>>,
availability: WorkerAvailability,
shutdown_timeout: time::Duration,
) {
availability.set(false);
let mut wrk = MAX_CONNS_COUNTER.with(|conns| Worker {
@ -151,8 +155,7 @@ impl Worker {
.map_err(|e| {
error!("Can not start worker: {:?}", e);
Arbiter::current().do_send(StopArbiter(0));
})
.and_then(move |services| {
}).and_then(move |services| {
wrk.services.extend(services);
wrk
}),

8
src/service/mod.rs
View File

@ -27,7 +27,9 @@ pub trait ServiceExt: Service {
/// Apply function to specified service and use it as a next service in
/// chain.
fn apply<S, I, F, R>(
self, service: I, f: F,
self,
service: I,
f: F,
) -> AndThen<Self, Apply<S, F, R, Self::Response>>
where
Self: Sized,
@ -120,7 +122,9 @@ pub trait ServiceExt: Service {
pub trait NewServiceExt: NewService {
fn apply<S, I, F, R>(
self, service: I, f: F,
self,
service: I,
f: F,
) -> AndThenNewService<Self, ApplyNewService<S, F, R, Self::Response>>
where
Self: Sized,

16
src/timeout.rs
View File

@ -1,4 +1,4 @@
//! Tower middleware that applies a timeout to requests.
//! Service that applies a timeout to requests.
//!
//! If the response does not complete within the specified timeout, the response
//! will be aborted.
@ -34,13 +34,6 @@ impl<E: fmt::Debug> fmt::Debug for TimeoutError<E> {
}
}
/// `Timeout` response future
#[derive(Debug)]
pub struct TimeoutFut<T: NewService> {
fut: T::Future,
timeout: Duration,
}
impl<T> Timeout<T>
where
T: NewService + Clone,
@ -69,6 +62,13 @@ where
}
}
/// `Timeout` response future
#[derive(Debug)]
pub struct TimeoutFut<T: NewService> {
fut: T::Future,
timeout: Duration,
}
impl<T> Future for TimeoutFut<T>
where
T: NewService,