actix/actix-net
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-net synced 2025-01-31 12:42:09 +01:00
Code Issues Releases Wiki Activity

refactor Connector service

Browse Source
This commit is contained in:
Nikolay Kim 2018-08-29 15:15:24 -07:00
parent 7ec92f7b1c
commit 361ffd8d2f
3 changed files with 121 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

141
src/connector.rs
View File

@ -1,5 +1,6 @@
use std::collections::VecDeque; use std::collections::VecDeque;
use std::io; use std::io;
use std::marker::PhantomData;
use std::net::SocketAddr; use std::net::SocketAddr;
use futures::{ use futures::{
@ -15,6 +16,16 @@ use trust_dns_resolver::{AsyncResolver, Background};
use super::{NewService, Service}; use super::{NewService, Service};
pub trait HostAware {
fn host(&self) -> &str;
}
impl HostAware for String {
fn host(&self) -> &str {
self.as_ref()
}
}
#[derive(Fail, Debug)] #[derive(Fail, Debug)]
pub enum ConnectorError { pub enum ConnectorError {
/// Failed to resolve the hostname /// Failed to resolve the hostname
@ -35,11 +46,12 @@ pub struct ConnectionInfo {
pub addr: SocketAddr, pub addr: SocketAddr,
} }
pub struct Connector { pub struct Connector<T> {
resolver: AsyncResolver, resolver: AsyncResolver,
req: PhantomData<T>,
} }
impl Default for Connector { impl<T: HostAware> Default for Connector<T> {
fn default() -> Self { fn default() -> Self {
let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() { let (cfg, opts) = if let Ok((cfg, opts)) = read_system_conf() {
(cfg, opts) (cfg, opts)
@ -51,66 +63,72 @@ impl Default for Connector {
} }
} }
impl Connector { impl<T: HostAware> Connector<T> {
pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self { pub fn new(cfg: ResolverConfig, opts: ResolverOpts) -> Self {
let (resolver, bg) = AsyncResolver::new(cfg, opts); let (resolver, bg) = AsyncResolver::new(cfg, opts);
tokio::spawn(bg); tokio::spawn(bg);
Connector { resolver } Connector {
resolver,
req: PhantomData,
}
} }
pub fn new_service<E>() -> impl NewService< pub fn new_service<E>() -> impl NewService<
Request = String, Request = T,
Response = (ConnectionInfo, TcpStream), Response = (T, ConnectionInfo, TcpStream),
Error = ConnectorError, Error = ConnectorError,
InitError = E, InitError = E,
> + Clone { > + Clone {
|| -> FutureResult<Connector, E> { ok(Connector::default()) } || -> FutureResult<Connector<T>, E> { ok(Connector::default()) }
} }
pub fn new_service_with_config<E>( pub fn new_service_with_config<E>(
cfg: ResolverConfig, opts: ResolverOpts, cfg: ResolverConfig, opts: ResolverOpts,
) -> impl NewService< ) -> impl NewService<
Request = String, Request = T,
Response = (ConnectionInfo, TcpStream), Response = (T, ConnectionInfo, TcpStream),
Error = ConnectorError, Error = ConnectorError,
InitError = E, InitError = E,
> + Clone { > + Clone {
move || -> FutureResult<Connector, E> { ok(Connector::new(cfg.clone(), opts.clone())) } move || -> FutureResult<Connector<T>, E> {
} ok(Connector::new(cfg.clone(), opts.clone()))
}
impl Clone for Connector {
fn clone(&self) -> Self {
Connector {
resolver: self.resolver.clone(),
} }
} }
} }
impl Service for Connector { impl<T> Clone for Connector<T> {
type Request = String; fn clone(&self) -> Self {
type Response = (ConnectionInfo, TcpStream); Connector {
resolver: self.resolver.clone(),
req: PhantomData,
}
}
}
impl<T: HostAware> Service for Connector<T> {
type Request = T;
type Response = (T, ConnectionInfo, TcpStream);
type Error = ConnectorError; type Error = ConnectorError;
type Future = ConnectorFuture; type Future = ConnectorFuture<T>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> { fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(())) Ok(Async::Ready(()))
} }
fn call(&mut self, addr: String) -> Self::Future { fn call(&mut self, req: Self::Request) -> Self::Future {
let fut = ResolveFut::new(addr, 0, &self.resolver); let fut = ResolveFut::new(req, 0, &self.resolver);
ConnectorFuture { fut, fut2: None } ConnectorFuture { fut, fut2: None }
} }
} }
pub struct ConnectorFuture { pub struct ConnectorFuture<T: HostAware> {
fut: ResolveFut, fut: ResolveFut<T>,
fut2: Option<TcpConnector>, fut2: Option<TcpConnector<T>>,
} }
impl Future for ConnectorFuture { impl<T: HostAware> Future for ConnectorFuture<T> {
type Item = (ConnectionInfo, TcpStream); type Item = (T, ConnectionInfo, TcpStream);
type Error = ConnectorError; type Error = ConnectorError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> { fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
@ -118,8 +136,8 @@ impl Future for ConnectorFuture {
return fut.poll(); return fut.poll();
} }
match self.fut.poll()? { match self.fut.poll()? {
Async::Ready((host, addrs)) => { Async::Ready((req, host, addrs)) => {
self.fut2 = Some(TcpConnector::new(host, addrs)); self.fut2 = Some(TcpConnector::new(req, host, addrs));
self.poll() self.poll()
} }
Async::NotReady => Ok(Async::NotReady), Async::NotReady => Ok(Async::NotReady),
@ -128,7 +146,8 @@ impl Future for ConnectorFuture {
} }
/// Resolver future /// Resolver future
struct ResolveFut { struct ResolveFut<T> {
req: Option<T>,
host: Option<String>, host: Option<String>,
port: u16, port: u16,
lookup: Option<Background<LookupIpFuture>>, lookup: Option<Background<LookupIpFuture>>,
@ -137,20 +156,25 @@ struct ResolveFut {
error2: Option<String>, error2: Option<String>,
} }
impl ResolveFut { impl<T: HostAware> ResolveFut<T> {
pub fn new(addr: String, port: u16, resolver: &AsyncResolver) -> Self { pub fn new(addr: T, port: u16, resolver: &AsyncResolver) -> Self {
// we need to do dns resolution // we need to do dns resolution
match ResolveFut::parse(addr.as_ref(), port) { match ResolveFut::<T>::parse(addr.host(), port) {
Ok((host, port)) => ResolveFut { Ok((host, port)) => {
port, let lookup = Some(resolver.lookup_ip(host.as_str()));
host: Some(host.to_owned()), ResolveFut {
lookup: Some(resolver.lookup_ip(host)), port,
addrs: None, lookup,
error: None, req: Some(addr),
error2: None, host: Some(host),
}, addrs: None,
error: None,
error2: None,
}
}
Err(err) => ResolveFut { Err(err) => ResolveFut {
port, port,
req: None,
host: None, host: None,
lookup: None, lookup: None,
addrs: None, addrs: None,
@ -160,7 +184,7 @@ impl ResolveFut {
} }
} }
fn parse(addr: &str, port: u16) -> Result<(&str, u16), ConnectorError> { fn parse(addr: &str, port: u16) -> Result<(String, u16), ConnectorError> {
macro_rules! try_opt { macro_rules! try_opt {
($e:expr, $msg:expr) => { ($e:expr, $msg:expr) => {
match $e { match $e {
@ -176,12 +200,12 @@ impl ResolveFut {
let port_str = parts_iter.next().unwrap_or(""); let port_str = parts_iter.next().unwrap_or("");
let port: u16 = port_str.parse().unwrap_or(port); let port: u16 = port_str.parse().unwrap_or(port);
Ok((host, port)) Ok((host.to_owned(), port))
} }
} }
impl Future for ResolveFut { impl<T> Future for ResolveFut<T> {
type Item = (String, VecDeque<SocketAddr>); type Item = (T, String, VecDeque<SocketAddr>);
type Error = ConnectorError; type Error = ConnectorError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> { fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
@ -190,7 +214,11 @@ impl Future for ResolveFut {
} else if let Some(err) = self.error2.take() { } else if let Some(err) = self.error2.take() {
Err(ConnectorError::Resolver(err)) Err(ConnectorError::Resolver(err))
} else if let Some(addrs) = self.addrs.take() { } else if let Some(addrs) = self.addrs.take() {
Ok(Async::Ready((self.host.take().unwrap(), addrs))) Ok(Async::Ready((
self.req.take().unwrap(),
self.host.take().unwrap(),
addrs,
)))
} else { } else {
match self.lookup.as_mut().unwrap().poll() { match self.lookup.as_mut().unwrap().poll() {
Ok(Async::NotReady) => Ok(Async::NotReady), Ok(Async::NotReady) => Ok(Async::NotReady),
@ -204,7 +232,11 @@ impl Future for ResolveFut {
"Expect at least one A dns record".to_owned(), "Expect at least one A dns record".to_owned(),
)) ))
} else { } else {
Ok(Async::Ready((self.host.take().unwrap(), addrs))) Ok(Async::Ready((
self.req.take().unwrap(),
self.host.take().unwrap(),
addrs,
)))
} }
} }
Err(err) => Err(ConnectorError::Resolver(format!("{}", err))), Err(err) => Err(ConnectorError::Resolver(format!("{}", err))),
@ -214,17 +246,19 @@ impl Future for ResolveFut {
} }
/// Tcp stream connector /// Tcp stream connector
pub struct TcpConnector { pub struct TcpConnector<T> {
req: Option<T>,
host: Option<String>, host: Option<String>,
addr: Option<SocketAddr>, addr: Option<SocketAddr>,
addrs: VecDeque<SocketAddr>, addrs: VecDeque<SocketAddr>,
stream: Option<ConnectFuture>, stream: Option<ConnectFuture>,
} }
impl TcpConnector { impl<T> TcpConnector<T> {
pub fn new(host: String, addrs: VecDeque<SocketAddr>) -> TcpConnector { pub fn new(req: T, host: String, addrs: VecDeque<SocketAddr>) -> TcpConnector<T> {
TcpConnector { TcpConnector {
addrs, addrs,
req: Some(req),
host: Some(host), host: Some(host),
addr: None, addr: None,
stream: None, stream: None,
@ -232,8 +266,8 @@ impl TcpConnector {
} }
} }
impl Future for TcpConnector { impl<T> Future for TcpConnector<T> {
type Item = (ConnectionInfo, TcpStream); type Item = (T, ConnectionInfo, TcpStream);
type Error = ConnectorError; type Error = ConnectorError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> { fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
@ -243,6 +277,7 @@ impl Future for TcpConnector {
match new.poll() { match new.poll() {
Ok(Async::Ready(sock)) => { Ok(Async::Ready(sock)) => {
return Ok(Async::Ready(( return Ok(Async::Ready((
self.req.take().unwrap(),
ConnectionInfo { ConnectionInfo {
host: self.host.take().unwrap(), host: self.host.take().unwrap(),
addr: self.addr.take().unwrap(), addr: self.addr.take().unwrap(),

3
src/lib.rs
View File

@ -53,7 +53,7 @@ pub use tower_service::{NewService, Service};
pub(crate) mod accept; pub(crate) mod accept;
pub mod configurable; pub mod configurable;
mod connector; pub mod connector;
mod server; mod server;
mod server_service; mod server_service;
pub mod service; pub mod service;
@ -61,7 +61,6 @@ pub mod ssl;
mod worker; mod worker;
pub use configurable::{IntoNewConfigurableService, NewConfigurableService}; pub use configurable::{IntoNewConfigurableService, NewConfigurableService};
pub use connector::{ConnectionInfo, Connector, ConnectorError};
pub use server::Server; pub use server::Server;
pub use service::{IntoNewService, IntoService, NewServiceExt, ServiceExt}; pub use service::{IntoNewService, IntoService, NewServiceExt, ServiceExt};

53
src/ssl/openssl.rs
View File

@ -92,83 +92,94 @@ impl<T: AsyncRead + AsyncWrite> Service for OpensslAcceptorService<T> {
} }
/// Openssl connector factory /// Openssl connector factory
pub struct OpensslConnector<T> { pub struct OpensslConnector<T, Io> {
connector: SslConnector, connector: SslConnector,
io: PhantomData<T>, t: PhantomData<T>,
io: PhantomData<Io>,
} }
impl<T> OpensslConnector<T> { impl<T, Io> OpensslConnector<T, Io> {
pub fn new(connector: SslConnector) -> Self { pub fn new(connector: SslConnector) -> Self {
OpensslConnector { OpensslConnector {
connector, connector,
t: PhantomData,
io: PhantomData, io: PhantomData,
} }
} }
} }
impl<T> Clone for OpensslConnector<T> { impl<T, Io> Clone for OpensslConnector<T, Io> {
fn clone(&self) -> Self { fn clone(&self) -> Self {
Self { Self {
connector: self.connector.clone(), connector: self.connector.clone(),
t: PhantomData,
io: PhantomData, io: PhantomData,
} }
} }
} }
impl<T: AsyncRead + AsyncWrite> NewService for OpensslConnector<T> { impl<T, Io: AsyncRead + AsyncWrite> NewService for OpensslConnector<T, Io> {
type Request = (ConnectionInfo, T); type Request = (T, ConnectionInfo, Io);
type Response = (ConnectionInfo, SslStream<T>); type Response = (T, ConnectionInfo, SslStream<Io>);
type Error = Error; type Error = Error;
type Service = OpensslConnectorService<T>; type Service = OpensslConnectorService<T, Io>;
type InitError = io::Error; type InitError = io::Error;
type Future = FutureResult<Self::Service, Self::InitError>; type Future = FutureResult<Self::Service, Self::InitError>;
fn new_service(&self) -> Self::Future { fn new_service(&self) -> Self::Future {
future::ok(OpensslConnectorService { future::ok(OpensslConnectorService {
connector: self.connector.clone(), connector: self.connector.clone(),
t: PhantomData,
io: PhantomData, io: PhantomData,
}) })
} }
} }
pub struct OpensslConnectorService<T> { pub struct OpensslConnectorService<T, Io> {
connector: SslConnector, connector: SslConnector,
io: PhantomData<T>, t: PhantomData<T>,
io: PhantomData<Io>,
} }
impl<T: AsyncRead + AsyncWrite> Service for OpensslConnectorService<T> { impl<T, Io: AsyncRead + AsyncWrite> Service for OpensslConnectorService<T, Io> {
type Request = (ConnectionInfo, T); type Request = (T, ConnectionInfo, Io);
type Response = (ConnectionInfo, SslStream<T>); type Response = (T, ConnectionInfo, SslStream<Io>);
type Error = Error; type Error = Error;
type Future = ConnectAsyncExt<T>; type Future = ConnectAsyncExt<T, Io>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> { fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(())) Ok(Async::Ready(()))
} }
fn call(&mut self, (info, stream): Self::Request) -> Self::Future { fn call(&mut self, (req, info, stream): Self::Request) -> Self::Future {
ConnectAsyncExt { ConnectAsyncExt {
fut: SslConnectorExt::connect_async(&self.connector, &info.host, stream), fut: SslConnectorExt::connect_async(&self.connector, &info.host, stream),
req: Some(req),
host: Some(info), host: Some(info),
} }
} }
} }
pub struct ConnectAsyncExt<T> { pub struct ConnectAsyncExt<T, Io> {
fut: ConnectAsync<T>, fut: ConnectAsync<Io>,
req: Option<T>,
host: Option<ConnectionInfo>, host: Option<ConnectionInfo>,
} }
impl<T> Future for ConnectAsyncExt<T> impl<T, Io> Future for ConnectAsyncExt<T, Io>
where where
T: AsyncRead + AsyncWrite, Io: AsyncRead + AsyncWrite,
{ {
type Item = (ConnectionInfo, SslStream<T>); type Item = (T, ConnectionInfo, SslStream<Io>);
type Error = Error; type Error = Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> { fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.fut.poll()? { match self.fut.poll()? {
Async::Ready(stream) => Ok(Async::Ready((self.host.take().unwrap(), stream))), Async::Ready(stream) => Ok(Async::Ready((
self.req.take().unwrap(),
self.host.take().unwrap(),
stream,
))),
Async::NotReady => Ok(Async::NotReady), Async::NotReady => Ok(Async::NotReady),
} }
} }