Merge pull request #672 from sfackler/tokio-03

Upgrade to tokio 0.3

Author: sfackler

.circleci/config.yml

@@ -22,7 +22,7 @@ version: 2
 jobs:
   build:
     docker:
-      - image: rust:1.41.0
+      - image: rust:1.45.0
         environment:
           RUSTFLAGS: -D warnings
       - image: sfackler/rust-postgres-test:6

codegen/src/type_gen.rs

@@ -136,10 +136,7 @@ impl<'a> DatParser<'a> {
     fn peek(&mut self, target: char) -> bool {
         self.skip_ws();
 
-        match self.it.peek() {
-            Some((_, ch)) if *ch == target => true,
-            _ => false,
-        }
+        matches!(self.it.peek(), Some((_, ch)) if *ch == target)
     }
 
     fn eof(&mut self) {

postgres-native-tls/Cargo.toml

@@ -16,13 +16,12 @@ default = ["runtime"]
 runtime = ["tokio-postgres/runtime"]
 
 [dependencies]
-bytes = "0.5"
 futures = "0.3"
 native-tls = "0.2"
-tokio = "0.2"
-tokio-tls = "0.3"
+tokio = "0.3"
+tokio-native-tls = "0.2"
 tokio-postgres = { version = "0.5.0", path = "../tokio-postgres", default-features = false }
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "0.3", features = ["full"] }
 postgres = { version = "0.17.0", path = "../postgres" }

postgres-native-tls/src/lib.rs

@@ -48,13 +48,11 @@
 #![doc(html_root_url = "https://docs.rs/postgres-native-tls/0.3")]
 #![warn(rust_2018_idioms, clippy::all, missing_docs)]
 
-use bytes::{Buf, BufMut};
 use std::future::Future;
 use std::io;
-use std::mem::MaybeUninit;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio_postgres::tls;
 #[cfg(feature = "runtime")]
 use tokio_postgres::tls::MakeTlsConnect;
@@ -94,15 +92,15 @@ where
 
 /// A `TlsConnect` implementation using the `native-tls` crate.
 pub struct TlsConnector {
-    connector: tokio_tls::TlsConnector,
+    connector: tokio_native_tls::TlsConnector,
     domain: String,
 }
 
 impl TlsConnector {
     /// Creates a new connector configured to connect to the specified domain.
     pub fn new(connector: native_tls::TlsConnector, domain: &str) -> TlsConnector {
         TlsConnector {
-            connector: tokio_tls::TlsConnector::from(connector),
+            connector: tokio_native_tls::TlsConnector::from(connector),
             domain: domain.to_string(),
         }
     }
@@ -129,34 +127,19 @@ where
 }
 
 /// The stream returned by `TlsConnector`.
-pub struct TlsStream<S>(tokio_tls::TlsStream<S>);
+pub struct TlsStream<S>(tokio_native_tls::TlsStream<S>);
 
 impl<S> AsyncRead for TlsStream<S>
 where
     S: AsyncRead + AsyncWrite + Unpin,
 {
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
-        self.0.prepare_uninitialized_buffer(buf)
-    }
-
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
         Pin::new(&mut self.0).poll_read(cx, buf)
     }
-
-    fn poll_read_buf<B: BufMut>(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut B,
-    ) -> Poll<io::Result<usize>>
-    where
-        Self: Sized,
-    {
-        Pin::new(&mut self.0).poll_read_buf(cx, buf)
-    }
 }
 
 impl<S> AsyncWrite for TlsStream<S>
@@ -178,25 +161,16 @@ where
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         Pin::new(&mut self.0).poll_shutdown(cx)
     }
-
-    fn poll_write_buf<B: Buf>(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut B,
-    ) -> Poll<io::Result<usize>>
-    where
-        Self: Sized,
-    {
-        Pin::new(&mut self.0).poll_write_buf(cx, buf)
-    }
 }
 
 impl<S> tls::TlsStream for TlsStream<S>
 where
     S: AsyncRead + AsyncWrite + Unpin,
 {
     fn channel_binding(&self) -> ChannelBinding {
-        // FIXME https://github.com/tokio-rs/tokio/issues/1383
-        ChannelBinding::none()
+        match self.0.get_ref().tls_server_end_point().ok().flatten() {
+            Some(buf) => ChannelBinding::tls_server_end_point(buf),
+            None => ChannelBinding::none(),
+        }
     }
 }

postgres-openssl/Cargo.toml

@@ -16,13 +16,12 @@ default = ["runtime"]
 runtime = ["tokio-postgres/runtime"]
 
 [dependencies]
-bytes = "0.5"
 futures = "0.3"
 openssl = "0.10"
-tokio = "0.2"
-tokio-openssl = "0.4"
+tokio = "0.3"
+tokio-openssl = "0.5"
 tokio-postgres = { version = "0.5.0", path = "../tokio-postgres", default-features = false }
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["full"] }
+tokio = { version = "0.3", features = ["full"] }
 postgres = { version = "0.17.0", path = "../postgres" }

postgres-openssl/src/lib.rs

@@ -42,7 +42,6 @@
 #![doc(html_root_url = "https://docs.rs/postgres-openssl/0.3")]
 #![warn(rust_2018_idioms, clippy::all, missing_docs)]
 
-use bytes::{Buf, BufMut};
 #[cfg(feature = "runtime")]
 use openssl::error::ErrorStack;
 use openssl::hash::MessageDigest;
@@ -53,12 +52,11 @@ use openssl::ssl::{ConnectConfiguration, SslRef};
 use std::fmt::Debug;
 use std::future::Future;
 use std::io;
-use std::mem::MaybeUninit;
 use std::pin::Pin;
 #[cfg(feature = "runtime")]
 use std::sync::Arc;
 use std::task::{Context, Poll};
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio_openssl::{HandshakeError, SslStream};
 use tokio_postgres::tls;
 #[cfg(feature = "runtime")]
@@ -157,28 +155,13 @@ impl<S> AsyncRead for TlsStream<S>
 where
     S: AsyncRead + AsyncWrite + Unpin,
 {
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [MaybeUninit<u8>]) -> bool {
-        self.0.prepare_uninitialized_buffer(buf)
-    }
-
     fn poll_read(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
         Pin::new(&mut self.0).poll_read(cx, buf)
     }
-
-    fn poll_read_buf<B: BufMut>(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut B,
-    ) -> Poll<io::Result<usize>>
-    where
-        Self: Sized,
-    {
-        Pin::new(&mut self.0).poll_read_buf(cx, buf)
-    }
 }
 
 impl<S> AsyncWrite for TlsStream<S>
@@ -200,17 +183,6 @@ where
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         Pin::new(&mut self.0).poll_shutdown(cx)
     }
-
-    fn poll_write_buf<B: Buf>(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut B,
-    ) -> Poll<io::Result<usize>>
-    where
-        Self: Sized,
-    {
-        Pin::new(&mut self.0).poll_write_buf(cx, buf)
-    }
 }
 
 impl<S> tls::TlsStream for TlsStream<S>

postgres-protocol/src/authentication/sasl.rs

@@ -330,10 +330,7 @@ impl<'a> Parser<'a> {
     }
 
     fn printable(&mut self) -> io::Result<&'a str> {
-        self.take_while(|c| match c {
-            '\x21'..='\x2b' | '\x2d'..='\x7e' => true,
-            _ => false,
-        })
+        self.take_while(|c| matches!(c, '\x21'..='\x2b' | '\x2d'..='\x7e'))
     }
 
     fn nonce(&mut self) -> io::Result<&'a str> {
@@ -343,10 +340,7 @@ impl<'a> Parser<'a> {
     }
 
     fn base64(&mut self) -> io::Result<&'a str> {
-        self.take_while(|c| match c {
-            'a'..='z' | 'A'..='Z' | '0'..='9' | '/' | '+' | '=' => true,
-            _ => false,
-        })
+        self.take_while(|c| matches!(c, 'a'..='z' | 'A'..='Z' | '0'..='9' | '/' | '+' | '='))
     }
 
     fn salt(&mut self) -> io::Result<&'a str> {
@@ -356,10 +350,7 @@ impl<'a> Parser<'a> {
     }
 
     fn posit_number(&mut self) -> io::Result<u32> {
-        let n = self.take_while(|c| match c {
-            '0'..='9' => true,
-            _ => false,
-        })?;
+        let n = self.take_while(|c| matches!(c, '0'..='9'))?;
         n.parse()
             .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))
     }
@@ -396,10 +387,7 @@ impl<'a> Parser<'a> {
     }
 
     fn value(&mut self) -> io::Result<&'a str> {
-        self.take_while(|c| match c {
-            '\0' | '=' | ',' => false,
-            _ => true,
-        })
+        self.take_while(|c| matches!(c, '\0' | '=' | ','))
     }
 
     fn server_error(&mut self) -> io::Result<Option<&'a str>> {

postgres-types/src/lib.rs

@@ -144,10 +144,7 @@ const NSEC_PER_USEC: u64 = 1_000;
 macro_rules! accepts {
     ($($expected:ident),+) => (
         fn accepts(ty: &$crate::Type) -> bool {
-            match *ty {
-                $($crate::Type::$expected)|+ => true,
-                _ => false
-            }
+            matches!(*ty, $($crate::Type::$expected)|+)
         }
     )
 }

postgres-types/src/special.rs

@@ -75,10 +75,7 @@ impl<'a, T: FromSql<'a>> FromSql<'a> for Timestamp<T> {
     }
 
     fn accepts(ty: &Type) -> bool {
-        match *ty {
-            Type::TIMESTAMP | Type::TIMESTAMPTZ if T::accepts(ty) => true,
-            _ => false,
-        }
+        matches!(*ty, Type::TIMESTAMP | Type::TIMESTAMPTZ if T::accepts(ty))
     }
 }
 
@@ -99,10 +96,7 @@ impl<T: ToSql> ToSql for Timestamp<T> {
     }
 
     fn accepts(ty: &Type) -> bool {
-        match *ty {
-            Type::TIMESTAMP | Type::TIMESTAMPTZ if T::accepts(ty) => true,
-            _ => false,
-        }
+        matches!(*ty, Type::TIMESTAMP | Type::TIMESTAMPTZ if T::accepts(ty))
     }
 
     to_sql_checked!();

postgres/Cargo.toml

@@ -36,7 +36,7 @@ fallible-iterator = "0.2"
 futures = "0.3"
 tokio-postgres = { version = "0.5.5", path = "../tokio-postgres" }
 
-tokio = { version = "0.2", features = ["rt-core", "time"] }
+tokio = { version = "0.3", features = ["rt", "time"] }
 log = "0.4"
 
 [dev-dependencies]

postgres/src/cancel_token.rs

@@ -26,9 +26,8 @@ impl CancelToken {
     where
         T: MakeTlsConnect<Socket>,
     {
-        runtime::Builder::new()
+        runtime::Builder::new_current_thread()
             .enable_all()
-            .basic_scheduler()
             .build()
             .unwrap() // FIXME don't unwrap
             .block_on(self.0.cancel_query(tls))

postgres/src/config.rs

@@ -336,9 +336,8 @@ impl Config {
         T::Stream: Send,
         <T::TlsConnect as TlsConnect<Socket>>::Future: Send,
     {
-        let mut runtime = runtime::Builder::new()
+        let runtime = runtime::Builder::new_current_thread()
             .enable_all()
-            .basic_scheduler()
             .build()
             .unwrap(); // FIXME don't unwrap
 

postgres/src/connection.rs

@@ -45,7 +45,8 @@ impl Connection {
     where
         F: FnOnce() -> T,
     {
-        self.runtime.enter(f)
+        let _guard = self.runtime.enter();
+        f()
     }
 
     pub fn block_on<F, T>(&mut self, future: F) -> Result<T, Error>

postgres/src/notifications.rs

@@ -6,7 +6,7 @@ use fallible_iterator::FallibleIterator;
 use futures::{ready, FutureExt};
 use std::task::Poll;
 use std::time::Duration;
-use tokio::time::{self, Delay, Instant};
+use tokio::time::{self, Instant, Sleep};
 
 /// Notifications from a PostgreSQL backend.
 pub struct Notifications<'a> {
@@ -64,7 +64,7 @@ impl<'a> Notifications<'a> {
     /// This iterator may start returning `Some` after previously returning `None` if more notifications are received.
     pub fn timeout_iter(&mut self, timeout: Duration) -> TimeoutIter<'_> {
         TimeoutIter {
-            delay: self.connection.enter(|| time::delay_for(timeout)),
+            delay: self.connection.enter(|| time::sleep(timeout)),
             timeout,
             connection: self.connection.as_ref(),
         }
@@ -124,7 +124,7 @@ impl<'a> FallibleIterator for BlockingIter<'a> {
 /// A time-limited blocking iterator over pending notifications.
 pub struct TimeoutIter<'a> {
     connection: ConnectionRef<'a>,
-    delay: Delay,
+    delay: Sleep,
     timeout: Duration,
 }