Worked around disjoint AsyncRead/Write traits between tokio and

futures::io See tokio-rs/tokio#1297 and async-rs/async-std#54
kevinastone · Nov 5, 2019 · ed39a7a · ed39a7a
1 parent da4835f
commit ed39a7a
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 38 deletions.
diff --git a/gotham/Cargo.toml b/gotham/Cargo.toml
@@ -50,6 +50,7 @@ httpdate = "0.3"
 failure = "0.1"
 futures-rustls = { git = "https://github.com/quininer/tokio-rustls", branch = "futures-rustls", optional = true }
 tokio-io = "=0.2.0-alpha.6"
+futures-tokio-compat = { git = "https://github.com/nemo157/futures-tokio-compat/" }
 
 [dev-dependencies]
 gotham_derive = "0.5.0-dev"

diff --git a/gotham/src/lib.rs b/gotham/src/lib.rs
@@ -91,10 +91,10 @@ pub fn bind_server<NH, F, Wrapped, Wrap>(
     listener: TcpListener,
     new_handler: NH,
     mut wrap: Wrap,
-) -> impl Future<Output = Result<(), ()>>
+) -> impl TryFuture<Ok = (), Error = ()>
 where
     NH: NewHandler + 'static,
-    F: Future<Output = Result<Wrapped, ()>> + Send + 'static,
+    F: Future<Output = Result<Wrapped, ()>> + Unpin + Send + 'static,
     Wrapped: Unpin + AsyncRead + AsyncWrite + Send + 'static,
     Wrap: FnMut(TcpStream) -> F,
 {
@@ -119,6 +119,6 @@ where
 
             executor::spawn(handler.map(|_| ()));
 
-            future::ok(())
+            future::ok(()).boxed()
         })
 }
diff --git a/gotham/src/tls.rs b/gotham/src/tls.rs
@@ -66,7 +66,7 @@ where
     A: ToSocketAddrs + 'static,
 {
     tcp_listener(addr)
-        // .map_err(|_| ())
+        .map_err(|_| ())
         .and_then(|listener| {
             let addr = listener.local_addr().unwrap();
 
@@ -76,25 +76,27 @@ where
             addr
             );
 
-            bind_server_rustls(listener, new_handler, tls_config)
+            bind_server_rustls(listener, new_handler, tls_config).map_err(|_| ())
         })
-        .then(|_| ()) // Ignore the result
+        .into_future()
+        .then(|_| future::ready(())) // Ignore the result
 }
 
 fn bind_server_rustls<NH>(
     listener: TcpListener,
     new_handler: NH,
     tls_config: rustls::ServerConfig,
-) -> impl TryFuture<Output = ()>
+) -> impl TryFuture<Ok = (), Error = ()>
 where
     NH: NewHandler + 'static,
 {
     let tls = TlsAcceptor::from(Arc::new(tls_config));
     bind_server(listener, new_handler, move |socket| {
-        tls.accept(socket).map_err(|e| {
-            error!(target: "gotham::tls", "TLS handshake error: {:?}", e);
-            ()
-        })
+        tls.accept(futures_tokio_compat::Compat::new(socket))
+            .map_ok(|stream| futures_tokio_compat::Compat::new(stream))
+            .map_err(|e| {
+                error!(target: "gotham::tls", "TLS handshake error: {:?}", e);
+                ()
+            })
     })
-    .then(|_| ())
 }
diff --git a/gotham/src/tls/test.rs b/gotham/src/tls/test.rs
@@ -167,7 +167,7 @@ impl TestServer {
             .runtime
             .write()
             .expect("unable to acquire read lock")
-            .spawn(fut);
+            .spawn(fut.into_future().map(|_| ()));
     }
 
     /// Returns a client connected to the `TestServer`. The transport is handled internally, and
@@ -212,7 +212,12 @@ pub struct TestConnect {
 }
 
 impl hyper::service::Service<Destination> for TestConnect {
-    type Response = (TcpStream, Connected);
+    type Response = (
+        futures_tokio_compat::Compat<
+            futures_rustls::client::TlsStream<futures_tokio_compat::Compat<TcpStream>>,
+        >,
+        Connected,
+    );
     type Error = CompatError;
     type Future =
         Pin<Box<dyn Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>>;
@@ -234,37 +239,19 @@ impl hyper::service::Service<Destination> for TestConnect {
         TcpStream::connect(&self.addr.clone())
             .and_then(move |stream: TcpStream| {
                 let domain = DNSNameRef::try_from_ascii_str(dst.host()).unwrap();
-                tls.connect(domain, stream)
+                tls.connect(domain, futures_tokio_compat::Compat::new(stream))
+                    .map_ok(|stream| futures_tokio_compat::Compat::new(stream))
             })
             .inspect_ok(|s| info!("Client TcpStream connected: {:?}", s))
             .map_ok(|s| (s, Connected::new()))
-            .map_err(|e| Error::from(e).compat())
+            .map_err(|e| {
+                info!("TLS TestClient error: {:?}", e);
+                Error::from(e).compat()
+            })
             .boxed()
     }
 }
 
-// TODO: Replace Connect
-// impl Connect for TestConnect {
-//     type Error = CompatError;
-
-//     fn connect(&self, dst: Destination) -> Self::Future {
-//         let tls = TlsConnector::from(self.config.clone());
-//         Box::new(
-//             TcpStream::connect(&self.addr)
-//                 .and_then(move |stream| {
-//                     let domain = DNSNameRef::try_from_ascii_str(dst.host()).unwrap();
-//                     tls.connect(domain, stream)
-//                 })
-//                 .inspect(|s| info!("Client TcpStream connected: {:?}", s))
-//                 .map(|s| (s, Connected::new()))
-//                 .map_err(|e| {
-//                     info!("TLS TestClient error: {:?}", e);
-//                     Error::from(e).compat()
-//                 }),
-//         )
-//     }
-// }
-
 #[cfg(test)]
 mod tests {
     use super::*;