Merge branch 'std-futures'

Author: sfackler

.circleci/config.yml

@@ -22,20 +22,20 @@ version: 2
 jobs:
   build:
     docker:
-      - image: rust:1.35.0
+      - image: rustlang/rust:nightly
         environment:
           RUSTFLAGS: -D warnings
       - image: sfackler/rust-postgres-test:5
     steps:
       - checkout
-      - run: rustup component add rustfmt clippy
+#      - run: rustup component add rustfmt clippy
       - *RESTORE_REGISTRY
       - run: cargo generate-lockfile
       - *SAVE_REGISTRY
       - run: rustc --version > ~/rust-version
       - *RESTORE_DEPS
-      - run: cargo fmt --all -- --check
-      - run: cargo clippy --all --all-targets --all-features
+#      - run: cargo fmt --all -- --check
+#      - run: cargo clippy --all --all-targets --all-features
       - run: cargo test --all
       - run: cargo test --manifest-path tokio-postgres/Cargo.toml --no-default-features
       - run: cargo test --manifest-path tokio-postgres/Cargo.toml --all-features

postgres-native-tls/Cargo.toml

@@ -16,12 +16,12 @@ default = ["runtime"]
 runtime = ["tokio-postgres/runtime"]
 
 [dependencies]
-futures = "0.1"
+futures-preview = "=0.3.0-alpha.18"
 native-tls = "0.2"
-tokio-io = "0.1"
-tokio-tls = "0.2.1"
+tokio-io = "=0.2.0-alpha.5"
+tokio-tls = "=0.3.0-alpha.5"
 tokio-postgres = { version = "0.4.0-rc.1", path = "../tokio-postgres", default-features = false }
 
 [dev-dependencies]
-tokio = "0.1.7"
+tokio = "=0.2.0-alpha.5"
 postgres = { version = "0.16.0-rc.1", path = "../postgres" }

postgres-native-tls/src/lib.rs

@@ -48,12 +48,13 @@
 #![doc(html_root_url = "https://docs.rs/postgres-native-tls/0.2.0-rc.1")]
 #![warn(rust_2018_idioms, clippy::all, missing_docs)]
 
-use futures::{try_ready, Async, Future, Poll};
+use std::future::Future;
+use std::pin::Pin;
 use tokio_io::{AsyncRead, AsyncWrite};
 #[cfg(feature = "runtime")]
 use tokio_postgres::tls::MakeTlsConnect;
 use tokio_postgres::tls::{ChannelBinding, TlsConnect};
-use tokio_tls::{Connect, TlsStream};
+use tokio_tls::TlsStream;
 
 #[cfg(test)]
 mod test;
@@ -76,7 +77,7 @@ impl MakeTlsConnector {
 #[cfg(feature = "runtime")]
 impl<S> MakeTlsConnect<S> for MakeTlsConnector
 where
-    S: AsyncRead + AsyncWrite,
+    S: AsyncRead + AsyncWrite + Unpin + 'static + Send,
 {
     type Stream = TlsStream<S>;
     type TlsConnect = TlsConnector;
@@ -105,35 +106,24 @@ impl TlsConnector {
 
 impl<S> TlsConnect<S> for TlsConnector
 where
-    S: AsyncRead + AsyncWrite,
+    S: AsyncRead + AsyncWrite + Unpin + 'static + Send,
 {
     type Stream = TlsStream<S>;
     type Error = native_tls::Error;
-    type Future = TlsConnectFuture<S>;
+    type Future = Pin<
+        Box<dyn Future<Output = Result<(TlsStream<S>, ChannelBinding), native_tls::Error>> + Send>,
+    >;
 
-    fn connect(self, stream: S) -> TlsConnectFuture<S> {
-        TlsConnectFuture(self.connector.connect(&self.domain, stream))
-    }
-}
-
-/// The future returned by `TlsConnector`.
-pub struct TlsConnectFuture<S>(Connect<S>);
-
-impl<S> Future for TlsConnectFuture<S>
-where
-    S: AsyncRead + AsyncWrite,
-{
-    type Item = (TlsStream<S>, ChannelBinding);
-    type Error = native_tls::Error;
+    fn connect(self, stream: S) -> Self::Future {
+        let future = async move {
+            let stream = self.connector.connect(&self.domain, stream).await?;
 
-    fn poll(&mut self) -> Poll<(TlsStream<S>, ChannelBinding), native_tls::Error> {
-        let stream = try_ready!(self.0.poll());
+            // FIXME https://github.com/tokio-rs/tokio/issues/1383
+            let channel_binding = ChannelBinding::none();
 
-        let channel_binding = match stream.get_ref().tls_server_end_point().unwrap_or(None) {
-            Some(buf) => ChannelBinding::tls_server_end_point(buf),
-            None => ChannelBinding::none(),
+            Ok((stream, channel_binding))
         };
 
-        Ok(Async::Ready((stream, channel_binding)))
+        Box::pin(future)
     }
 }

postgres-native-tls/src/test.rs

@@ -1,44 +1,40 @@
-use futures::{Future, Stream};
+use futures::{FutureExt, TryStreamExt};
 use native_tls::{self, Certificate};
 use tokio::net::TcpStream;
-use tokio::runtime::current_thread::Runtime;
 use tokio_postgres::tls::TlsConnect;
 
 #[cfg(feature = "runtime")]
 use crate::MakeTlsConnector;
 use crate::TlsConnector;
 
-fn smoke_test<T>(s: &str, tls: T)
+async fn smoke_test<T>(s: &str, tls: T)
 where
     T: TlsConnect<TcpStream>,
-    T::Stream: 'static,
+    T::Stream: 'static + Send,
 {
-    let mut runtime = Runtime::new().unwrap();
+    let stream = TcpStream::connect("127.0.0.1:5433")
+        .await
+        .unwrap();
 
     let builder = s.parse::<tokio_postgres::Config>().unwrap();
+    let (mut client, connection) = builder.connect_raw(stream, tls).await.unwrap();
 
-    let handshake = TcpStream::connect(&"127.0.0.1:5433".parse().unwrap())
-        .map_err(|e| panic!("{}", e))
-        .and_then(|s| builder.connect_raw(s, tls));
-    let (mut client, connection) = runtime.block_on(handshake).unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.spawn(connection);
+    let connection = connection.map(|r| r.unwrap());
+    tokio::spawn(connection);
 
-    let prepare = client.prepare("SELECT 1::INT4");
-    let statement = runtime.block_on(prepare).unwrap();
-    let select = client.query(&statement, &[]).collect().map(|rows| {
-        assert_eq!(rows.len(), 1);
-        assert_eq!(rows[0].get::<_, i32>(0), 1);
-    });
-    runtime.block_on(select).unwrap();
+    let stmt = client.prepare("SELECT $1::INT4").await.unwrap();
+    let rows = client
+        .query(&stmt, &[&1i32])
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap();
 
-    drop(statement);
-    drop(client);
-    runtime.run().unwrap();
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].get::<_, i32>(0), 1);
 }
 
-#[test]
-fn require() {
+#[tokio::test]
+async fn require() {
     let connector = native_tls::TlsConnector::builder()
         .add_root_certificate(
             Certificate::from_pem(include_bytes!("../../test/server.crt")).unwrap(),
@@ -48,11 +44,12 @@ fn require() {
     smoke_test(
         "user=ssl_user dbname=postgres sslmode=require",
         TlsConnector::new(connector, "localhost"),
-    );
+    )
+    .await;
 }
 
-#[test]
-fn prefer() {
+#[tokio::test]
+async fn prefer() {
     let connector = native_tls::TlsConnector::builder()
         .add_root_certificate(
             Certificate::from_pem(include_bytes!("../../test/server.crt")).unwrap(),
@@ -62,11 +59,12 @@ fn prefer() {
     smoke_test(
         "user=ssl_user dbname=postgres",
         TlsConnector::new(connector, "localhost"),
-    );
+    )
+    .await;
 }
 
-#[test]
-fn scram_user() {
+#[tokio::test]
+async fn scram_user() {
     let connector = native_tls::TlsConnector::builder()
         .add_root_certificate(
             Certificate::from_pem(include_bytes!("../../test/server.crt")).unwrap(),
@@ -76,14 +74,13 @@ fn scram_user() {
     smoke_test(
         "user=scram_user password=password dbname=postgres sslmode=require",
         TlsConnector::new(connector, "localhost"),
-    );
+    )
+    .await;
 }
 
-#[test]
+#[tokio::test]
 #[cfg(feature = "runtime")]
-fn runtime() {
-    let mut runtime = Runtime::new().unwrap();
-
+async fn runtime() {
     let connector = native_tls::TlsConnector::builder()
         .add_root_certificate(
             Certificate::from_pem(include_bytes!("../../test/server.crt")).unwrap(),
@@ -92,14 +89,22 @@ fn runtime() {
         .unwrap();
     let connector = MakeTlsConnector::new(connector);
 
-    let connect = tokio_postgres::connect(
+    let (mut client, connection) = tokio_postgres::connect(
         "host=localhost port=5433 user=postgres sslmode=require",
         connector,
-    );
-    let (mut client, connection) = runtime.block_on(connect).unwrap();
-    let connection = connection.map_err(|e| panic!("{}", e));
-    runtime.spawn(connection);
+    )
+    .await
+    .unwrap();
+    let connection = connection.map(|r| r.unwrap());
+    tokio::spawn(connection);
+
+    let stmt = client.prepare("SELECT $1::INT4").await.unwrap();
+    let rows = client
+        .query(&stmt, &[&1i32])
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap();
 
-    let execute = client.simple_query("SELECT 1").for_each(|_| Ok(()));
-    runtime.block_on(execute).unwrap();
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].get::<_, i32>(0), 1);
 }

postgres-openssl/Cargo.toml

@@ -16,12 +16,12 @@ default = ["runtime"]
 runtime = ["tokio-postgres/runtime"]
 
 [dependencies]
-futures = "0.1"
+futures-preview = "=0.3.0-alpha.18"
 openssl = "0.10"
-tokio-io = "0.1"
-tokio-openssl = "0.3"
+tokio-io = "=0.2.0-alpha.5"
+tokio-openssl = "=0.4.0-alpha.5"
 tokio-postgres = { version = "0.4.0-rc.1", path = "../tokio-postgres", default-features = false }
 
 [dev-dependencies]
-tokio = "0.1.7"
+tokio = "=0.2.0-alpha.5"
 postgres = { version = "0.16.0-rc.1", path = "../postgres" }

postgres-openssl/src/lib.rs

@@ -42,19 +42,20 @@
 #![doc(html_root_url = "https://docs.rs/postgres-openssl/0.2.0-rc.1")]
 #![warn(rust_2018_idioms, clippy::all, missing_docs)]
 
-use futures::{try_ready, Async, Future, Poll};
 #[cfg(feature = "runtime")]
 use openssl::error::ErrorStack;
 use openssl::hash::MessageDigest;
 use openssl::nid::Nid;
 #[cfg(feature = "runtime")]
 use openssl::ssl::SslConnector;
-use openssl::ssl::{ConnectConfiguration, HandshakeError, SslRef};
+use openssl::ssl::{ConnectConfiguration, SslRef};
 use std::fmt::Debug;
+use std::future::Future;
+use std::pin::Pin;
 #[cfg(feature = "runtime")]
 use std::sync::Arc;
 use tokio_io::{AsyncRead, AsyncWrite};
-use tokio_openssl::{ConnectAsync, ConnectConfigurationExt, SslStream};
+use tokio_openssl::{HandshakeError, SslStream};
 #[cfg(feature = "runtime")]
 use tokio_postgres::tls::MakeTlsConnect;
 use tokio_postgres::tls::{ChannelBinding, TlsConnect};
@@ -96,7 +97,7 @@ impl MakeTlsConnector {
 #[cfg(feature = "runtime")]
 impl<S> MakeTlsConnect<S> for MakeTlsConnector
 where
-    S: AsyncRead + AsyncWrite + Debug + 'static + Sync + Send,
+    S: AsyncRead + AsyncWrite + Unpin + Debug + 'static + Sync + Send,
 {
     type Stream = SslStream<S>;
     type TlsConnect = TlsConnector;
@@ -127,36 +128,27 @@ impl TlsConnector {
 
 impl<S> TlsConnect<S> for TlsConnector
 where
-    S: AsyncRead + AsyncWrite + Debug + 'static + Sync + Send,
+    S: AsyncRead + AsyncWrite + Unpin + Debug + 'static + Sync + Send,
 {
     type Stream = SslStream<S>;
     type Error = HandshakeError<S>;
-    type Future = TlsConnectFuture<S>;
+    type Future = Pin<
+        Box<dyn Future<Output = Result<(SslStream<S>, ChannelBinding), HandshakeError<S>>> + Send>,
+    >;
 
-    fn connect(self, stream: S) -> TlsConnectFuture<S> {
-        TlsConnectFuture(self.ssl.connect_async(&self.domain, stream))
-    }
-}
-
-/// The future returned by `TlsConnector`.
-pub struct TlsConnectFuture<S>(ConnectAsync<S>);
-
-impl<S> Future for TlsConnectFuture<S>
-where
-    S: AsyncRead + AsyncWrite + Debug + 'static + Sync + Send,
-{
-    type Item = (SslStream<S>, ChannelBinding);
-    type Error = HandshakeError<S>;
+    fn connect(self, stream: S) -> Self::Future {
+        let future = async move {
+            let stream = tokio_openssl::connect(self.ssl, &self.domain, stream).await?;
 
-    fn poll(&mut self) -> Poll<(SslStream<S>, ChannelBinding), HandshakeError<S>> {
-        let stream = try_ready!(self.0.poll());
+            let channel_binding = match tls_server_end_point(stream.ssl()) {
+                Some(buf) => ChannelBinding::tls_server_end_point(buf),
+                None => ChannelBinding::none(),
+            };
 
-        let channel_binding = match tls_server_end_point(stream.get_ref().ssl()) {
-            Some(buf) => ChannelBinding::tls_server_end_point(buf),
-            None => ChannelBinding::none(),
+            Ok((stream, channel_binding))
         };
 
-        Ok(Async::Ready((stream, channel_binding)))
+        Box::pin(future)
     }
 }