Updated to support tokio 0.2 and latest versions of all other deps

Author: jbg

Cargo.toml

@@ -1,19 +1,20 @@
 [package]
 name = "tokio-postgres-rustls"
 description = "Rustls integration for tokio-postgres"
-version = "0.2.0"
+version = "0.3.0"
 authors = ["jbg <jbg@dtmf.io>"]
 edition = "2018"
 license = "MIT"
 readme = "README.md"
 
 [dependencies]
-futures = "0.1.28"
-rustls = "0.15.2"
-tokio-io = "0.1.12"
-tokio-postgres = "0.4.0-rc.3"
-tokio-rustls = "0.10.0-alpha.3"
-webpki = "0.19.1"
+bytes = "0.5.3"
+futures = "0.3.1"
+rustls = "0.16.0"
+tokio = "0.2.6"
+tokio-postgres = "0.5.1"
+tokio-rustls = "0.12.1"
+webpki = "0.21.0"
 
 [dev-dependencies]
-tokio = "0.1.21"
+tokio = { version = "0.2.6", features = ["macros"] }

src/lib.rs

@@ -1,11 +1,18 @@
+#![feature(type_alias_impl_trait)]
+
 use std::{
     io,
+    future::Future,
+    mem::MaybeUninit,
+    pin::Pin,
     sync::Arc,
+    task::{Context, Poll},
 };
 
-use futures::Future;
+use bytes::{Buf, BufMut};
+use futures::future::TryFutureExt;
 use rustls::ClientConfig;
-use tokio_io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_postgres::tls::{ChannelBinding, MakeTlsConnect, TlsConnect};
 use tokio_rustls::{client::TlsStream, TlsConnector};
 use webpki::{DNSName, DNSNameRef};
@@ -23,13 +30,13 @@ impl MakeRustlsConnect {
 
 impl<S> MakeTlsConnect<S> for MakeRustlsConnect
 where
-    S: AsyncRead + AsyncWrite + Send + 'static
+    S: AsyncRead + AsyncWrite + Unpin,
 {
-    type Stream = TlsStream<S>;
+    type Stream = RustlsStream<S>;
     type TlsConnect = RustlsConnect;
-    type Error = io::Error;
+    type Error = std::io::Error;
 
-    fn make_tls_connect(&mut self, hostname: &str) -> Result<RustlsConnect, Self::Error> {
+    fn make_tls_connect(&mut self, hostname: &str) -> std::io::Result<RustlsConnect> {
         DNSNameRef::try_from_ascii_str(hostname)
             .map(|dns_name| RustlsConnect {
                 hostname: dns_name.to_owned(),
@@ -46,44 +53,84 @@ pub struct RustlsConnect {
 
 impl<S> TlsConnect<S> for RustlsConnect
 where
-    S: AsyncRead + AsyncWrite + Send + 'static
+    S: AsyncRead + AsyncWrite + Unpin,
 {
-    type Stream = TlsStream<S>;
-    type Error = io::Error;
-    type Future = Box<dyn Future<Item=(Self::Stream, ChannelBinding), Error=Self::Error> + Send>;
+    type Stream = RustlsStream<S>;
+    type Error = std::io::Error;
+    type Future = impl Future<Output = std::io::Result<RustlsStream<S>>>;
 
     fn connect(self, stream: S) -> Self::Future {
-        Box::new(
-            self.connector.connect(self.hostname.as_ref(), stream)
-                .map(|s| (s, ChannelBinding::none()))  // TODO
-        )
+        self.connector.connect(self.hostname.as_ref(), stream)
+            .map_ok(|s| RustlsStream(Box::pin(s)))
+    }
+}
+
+pub struct RustlsStream<S>(Pin<Box<TlsStream<S>>>);
+
+impl<S> tokio_postgres::tls::TlsStream for RustlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn channel_binding(&self) -> ChannelBinding {
+        ChannelBinding::none() // TODO
+    }
+}
+
+impl<S> AsyncRead for RustlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context, buf: &mut [u8]) -> Poll<tokio::io::Result<usize>> {
+        self.0.as_mut().poll_read(cx, buf)
+    }
+
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
+        self.0.prepare_uninitialized_buffer(buf)
+    }
+
+    fn poll_read_buf<B: BufMut>(mut self: Pin<&mut Self>, cx: &mut Context, buf: &mut B) -> Poll<tokio::io::Result<usize>>
+    where
+        Self: Sized,
+    {
+        self.0.as_mut().poll_read_buf(cx, buf)
+    }
+}
+
+impl<S> AsyncWrite for RustlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context, buf: &[u8]) -> Poll<tokio::io::Result<usize>> {
+        self.0.as_mut().poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<tokio::io::Result<()>> {
+        self.0.as_mut().poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<tokio::io::Result<()>> {
+        self.0.as_mut().poll_shutdown(cx)
+    }
+
+    fn poll_write_buf<B: Buf>(mut self: Pin<&mut Self>, cx: &mut Context, buf: &mut B) -> Poll<tokio::io::Result<usize>>
+    where
+        Self: Sized,
+    {
+        self.0.as_mut().poll_write_buf(cx, buf)
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use futures::{Future, Stream};
-    use tokio::runtime::current_thread;
+    use futures::future::TryFutureExt;
 
-    #[test]
-    fn it_works() {
+    #[tokio::test]
+    async fn it_works() {
         let config = rustls::ClientConfig::new();
         let tls = super::MakeRustlsConnect::new(config);
-        current_thread::block_on_all(
-            tokio_postgres::connect("sslmode=require host=localhost user=postgres", tls)
-                .map(|(client, connection)| {
-                    tokio::spawn(
-                        connection.map_err(|e| panic!("{:?}", e))
-                    );
-                    client
-                })
-                .and_then(|mut client| {
-                    client.prepare("SELECT 1")
-                        .map(|s| (client, s))
-                })
-                .and_then(|(mut client, statement)| {
-                    client.query(&statement, &[]).collect()
-                })
-        ).unwrap();
+        let (client, conn) = tokio_postgres::connect("sslmode=require host=localhost user=postgres", tls).await.unwrap();
+        tokio::spawn(conn.map_err(|e| panic!("{:?}", e)));
+        let stmt = client.prepare("SELECT 1").await.unwrap();
+        let _ = client.query(&stmt, &[]).await.unwrap();
     }
 }