add TcpStreamService

actix-connect/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.2.5] - 2019-09-05
+
+* Add `TcpConnectService`
+
 ## [0.2.4] - 2019-09-02
 
 * Use arbiter's storage for default async resolver

actix-connect/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-connect"
-version = "0.2.4"
+version = "0.2.5"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix Connector - tcp connector service"
 keywords = ["network", "framework", "async", "futures"]

actix-connect/src/lib.rs

@@ -28,7 +28,7 @@ pub use self::connect::{Address, Connect, Connection};
 pub use self::connector::{TcpConnector, TcpConnectorFactory};
 pub use self::error::ConnectError;
 pub use self::resolver::{Resolver, ResolverFactory};
-pub use self::service::{ConnectService, ConnectServiceFactory};
+pub use self::service::{ConnectService, ConnectServiceFactory, TcpConnectService};
 
 use actix_rt::Arbiter;
 use actix_service::{NewService, Service, ServiceExt};

actix-connect/src/service.rs

@@ -38,6 +38,14 @@ impl<T> ConnectServiceFactory<T> {
             resolver: self.resolver.service(),
         }
     }
+
+    /// Construct new tcp stream service
+    pub fn tcp_service(&self) -> TcpConnectService<T> {
+        TcpConnectService {
+            tcp: self.tcp.service(),
+            resolver: self.resolver.service(),
+        }
+    }
 }
 
 impl<T> Default for ConnectServiceFactory<T> {
@@ -121,3 +129,55 @@ impl<T: Address> Future for ConnectServiceResponse<T> {
         Ok(Async::NotReady)
     }
 }
+
+#[derive(Clone)]
+pub struct TcpConnectService<T> {
+    tcp: TcpConnector<T>,
+    resolver: Resolver<T>,
+}
+
+impl<T: Address> Service for TcpConnectService<T> {
+    type Request = Connect<T>;
+    type Response = TcpStream;
+    type Error = ConnectError;
+    type Future = TcpConnectServiceResponse<T>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        Ok(Async::Ready(()))
+    }
+
+    fn call(&mut self, req: Connect<T>) -> Self::Future {
+        TcpConnectServiceResponse {
+            fut1: Some(self.resolver.call(req)),
+            fut2: None,
+            tcp: self.tcp.clone(),
+        }
+    }
+}
+
+pub struct TcpConnectServiceResponse<T: Address> {
+    fut1: Option<<Resolver<T> as Service>::Future>,
+    fut2: Option<<TcpConnector<T> as Service>::Future>,
+    tcp: TcpConnector<T>,
+}
+
+impl<T: Address> Future for TcpConnectServiceResponse<T> {
+    type Item = TcpStream;
+    type Error = ConnectError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        if let Some(ref mut fut) = self.fut1 {
+            let res = try_ready!(fut.poll());
+            let _ = self.fut1.take();
+            self.fut2 = Some(self.tcp.call(res));
+        }
+
+        if let Some(ref mut fut) = self.fut2 {
+            if let Async::Ready(conn) = fut.poll()? {
+                return Ok(Async::Ready(conn.into_parts().0));
+            }
+        }
+
+        Ok(Async::NotReady)
+    }
+}