mirror of
https://github.com/fafhrd91/actix-net
synced 2024-11-27 18:02:58 +01:00
Connector::Response returns addr and tcp stream
This commit is contained in:
parent
3dbaef3ec1
commit
c69d675113
@ -43,7 +43,7 @@ fn main() {
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
let num = Arc::new(AtomicUsize::new(0));
|
let num = Arc::new(AtomicUsize::new(0));
|
||||||
let openssl = ssl::OpensslService::new(builder);
|
let openssl = ssl::OpensslAcceptor::new(builder);
|
||||||
|
|
||||||
// server start mutiple workers, it runs supplied `Fn` in each worker.
|
// server start mutiple workers, it runs supplied `Fn` in each worker.
|
||||||
Server::default()
|
Server::default()
|
||||||
@ -51,7 +51,10 @@ fn main() {
|
|||||||
let num = num.clone();
|
let num = num.clone();
|
||||||
|
|
||||||
// configure service
|
// configure service
|
||||||
openssl.clone().and_then((service, move || {
|
openssl
|
||||||
|
.clone()
|
||||||
|
.map_err(|e| io::Error::new(io::ErrorKind::Other, e))
|
||||||
|
.and_then((service, move || {
|
||||||
Ok::<_, io::Error>(ServiceState { num: num.clone() })
|
Ok::<_, io::Error>(ServiceState { num: num.clone() })
|
||||||
}))
|
}))
|
||||||
}).unwrap()
|
}).unwrap()
|
||||||
|
@ -235,7 +235,7 @@ where
|
|||||||
pub struct AndThenNewConfigurableService<A, B, C>
|
pub struct AndThenNewConfigurableService<A, B, C>
|
||||||
where
|
where
|
||||||
A: NewConfigurableService<C>,
|
A: NewConfigurableService<C>,
|
||||||
B: NewConfigurableService<C>
|
B: NewConfigurableService<C>,
|
||||||
{
|
{
|
||||||
a: A,
|
a: A,
|
||||||
b: B,
|
b: B,
|
||||||
@ -259,11 +259,7 @@ where
|
|||||||
|
|
||||||
impl<A, B, C> NewConfigurableService<C> for AndThenNewConfigurableService<A, B, C>
|
impl<A, B, C> NewConfigurableService<C> for AndThenNewConfigurableService<A, B, C>
|
||||||
where
|
where
|
||||||
A: NewConfigurableService<
|
A: NewConfigurableService<C, Response = B::Request, InitError = B::InitError>,
|
||||||
C,
|
|
||||||
Response = B::Request,
|
|
||||||
InitError = B::InitError,
|
|
||||||
>,
|
|
||||||
A::Error: Into<B::Error>,
|
A::Error: Into<B::Error>,
|
||||||
B: NewConfigurableService<C>,
|
B: NewConfigurableService<C>,
|
||||||
C: Clone,
|
C: Clone,
|
||||||
@ -359,7 +355,7 @@ where
|
|||||||
/// `MapErrNewService` new service combinator
|
/// `MapErrNewService` new service combinator
|
||||||
pub struct MapErrNewConfigurableService<A, F, E, C>
|
pub struct MapErrNewConfigurableService<A, F, E, C>
|
||||||
where
|
where
|
||||||
A: NewConfigurableService<C>
|
A: NewConfigurableService<C>,
|
||||||
{
|
{
|
||||||
a: A,
|
a: A,
|
||||||
f: F,
|
f: F,
|
||||||
@ -455,7 +451,7 @@ where
|
|||||||
/// `MapInitErr` service combinator
|
/// `MapInitErr` service combinator
|
||||||
pub struct MapInitErr<A, F, E, C>
|
pub struct MapInitErr<A, F, E, C>
|
||||||
where
|
where
|
||||||
A: NewConfigurableService<C>
|
A: NewConfigurableService<C>,
|
||||||
{
|
{
|
||||||
a: A,
|
a: A,
|
||||||
f: F,
|
f: F,
|
||||||
|
@ -51,7 +51,7 @@ impl Connector {
|
|||||||
|
|
||||||
impl Service for Connector {
|
impl Service for Connector {
|
||||||
type Request = String;
|
type Request = String;
|
||||||
type Response = TcpStream;
|
type Response = (String, TcpStream);
|
||||||
type Error = ConnectorError;
|
type Error = ConnectorError;
|
||||||
type Future = ConnectorFuture;
|
type Future = ConnectorFuture;
|
||||||
|
|
||||||
@ -60,25 +60,32 @@ impl Service for Connector {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn call(&mut self, addr: String) -> Self::Future {
|
fn call(&mut self, addr: String) -> Self::Future {
|
||||||
|
let fut = ResolveFut::new(&addr, 0, &self.resolver);
|
||||||
|
|
||||||
ConnectorFuture {
|
ConnectorFuture {
|
||||||
fut: ResolveFut::new(addr, 0, &self.resolver),
|
fut,
|
||||||
|
addr: Some(addr),
|
||||||
fut2: None,
|
fut2: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct ConnectorFuture {
|
pub struct ConnectorFuture {
|
||||||
|
addr: Option<String>,
|
||||||
fut: ResolveFut,
|
fut: ResolveFut,
|
||||||
fut2: Option<TcpConnector>,
|
fut2: Option<TcpConnector>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Future for ConnectorFuture {
|
impl Future for ConnectorFuture {
|
||||||
type Item = TcpStream;
|
type Item = (String, 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> {
|
||||||
if let Some(ref mut fut) = self.fut2 {
|
if let Some(ref mut fut) = self.fut2 {
|
||||||
return fut.poll();
|
return match fut.poll()? {
|
||||||
|
Async::Ready(stream) => Ok(Async::Ready((self.addr.take().unwrap(), stream))),
|
||||||
|
Async::NotReady => Ok(Async::NotReady),
|
||||||
|
};
|
||||||
}
|
}
|
||||||
match self.fut.poll()? {
|
match self.fut.poll()? {
|
||||||
Async::Ready(addrs) => {
|
Async::Ready(addrs) => {
|
||||||
@ -100,7 +107,7 @@ struct ResolveFut {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl ResolveFut {
|
impl ResolveFut {
|
||||||
pub fn new(addr: String, port: u16, resolver: &AsyncResolver) -> Self {
|
pub fn new(addr: &str, 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::parse(addr.as_ref(), port) {
|
||||||
Ok((host, port)) => ResolveFut {
|
Ok((host, port)) => ResolveFut {
|
||||||
|
@ -115,7 +115,7 @@ impl<T> Clone for OpensslConnector<T> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl<T: AsyncRead + AsyncWrite> NewService for OpensslConnector<T> {
|
impl<T: AsyncRead + AsyncWrite> NewService for OpensslConnector<T> {
|
||||||
type Request = T;
|
type Request = (String, T);
|
||||||
type Response = SslStream<T>;
|
type Response = SslStream<T>;
|
||||||
type Error = Error;
|
type Error = Error;
|
||||||
type Service = OpensslConnectorService<T>;
|
type Service = OpensslConnectorService<T>;
|
||||||
@ -130,17 +130,13 @@ impl<T: AsyncRead + AsyncWrite> NewService for OpensslConnector<T> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub trait OpensslDomain {
|
|
||||||
fn domain(&self) -> &str;
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct OpensslConnectorService<T> {
|
pub struct OpensslConnectorService<T> {
|
||||||
connector: SslConnector,
|
connector: SslConnector,
|
||||||
io: PhantomData<T>,
|
io: PhantomData<T>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T: AsyncRead + AsyncWrite> Service for OpensslConnectorService<T> {
|
impl<T: AsyncRead + AsyncWrite> Service for OpensslConnectorService<T> {
|
||||||
type Request = T;
|
type Request = (String, T);
|
||||||
type Response = SslStream<T>;
|
type Response = SslStream<T>;
|
||||||
type Error = Error;
|
type Error = Error;
|
||||||
type Future = ConnectAsync<T>;
|
type Future = ConnectAsync<T>;
|
||||||
@ -149,7 +145,7 @@ impl<T: AsyncRead + AsyncWrite> Service for OpensslConnectorService<T> {
|
|||||||
Ok(Async::Ready(()))
|
Ok(Async::Ready(()))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn call(&mut self, req: Self::Request) -> Self::Future {
|
fn call(&mut self, (host, stream): Self::Request) -> Self::Future {
|
||||||
SslConnectorExt::connect_async(&self.connector, "", req)
|
SslConnectorExt::connect_async(&self.connector, &host, stream)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user