Prefer sub-crates of futures

This removes `futures-executor` and `futures-io` from the transient dependencies.

Author: nickelc

Cargo.lock

@@ -15,7 +15,8 @@ description = "An async extension for Diesel the safe, extensible ORM and Query
 [dependencies]
 diesel = { version = "2.0.0", default-features = false, features = ["i-implement-a-third-party-backend-and-opt-into-breaking-changes"]}
 async-trait = "0.1.51"
-futures = "0.3.17"
+futures-channel = { version = "0.3.17", default-features = false, features = ["std", "sink"] }
+futures-util = { version = "0.3.17", default-features = false, features = ["std", "sink"] }
 tokio-postgres = { version = "0.7.2", optional = true}
 tokio = { version = "1", optional = true}
 mysql_async = { version = "0.30.0", optional = true}

Cargo.toml

@@ -70,7 +70,7 @@ use diesel::backend::Backend;
 use diesel::query_builder::{AsQuery, QueryFragment, QueryId};
 use diesel::row::Row;
 use diesel::{ConnectionResult, QueryResult};
-use futures::{Future, Stream};
+use futures_util::{Future, Stream};
 
 pub use scoped_futures;
 use scoped_futures::ScopedBoxFuture;

src/lib.rs

@@ -5,9 +5,9 @@ use diesel::mysql::{Mysql, MysqlType};
 use diesel::query_builder::{bind_collector::RawBytesBindCollector, QueryFragment, QueryId};
 use diesel::result::{ConnectionError, ConnectionResult};
 use diesel::QueryResult;
-use futures::future::BoxFuture;
-use futures::stream::BoxStream;
-use futures::{Future, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
+use futures_util::future::{self, BoxFuture};
+use futures_util::stream::{self, BoxStream};
+use futures_util::{Future, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
 use mysql_async::prelude::Queryable;
 use mysql_async::{Opts, OptsBuilder, Statement};
 
@@ -82,7 +82,7 @@ impl AsyncConnection for AsyncMysqlConnection {
                 _ => todo!(),
             };
 
-            let (tx, rx) = futures::channel::mpsc::channel(0);
+            let (tx, rx) = futures_channel::mpsc::channel(0);
 
             let yielder = async move {
                 let r = Self::poll_result_stream(conn, stmt_for_exec, binds, tx).await;
@@ -104,16 +104,15 @@ impl AsyncConnection for AsyncMysqlConnection {
                 r
             };
 
-            let fake_stream =
-                futures::stream::once(yielder).filter_map(|e: QueryResult<()>| async move {
-                    if let Err(e) = e {
-                        Some(Err(e))
-                    } else {
-                        None
-                    }
-                });
+            let fake_stream = stream::once(yielder).filter_map(|e: QueryResult<()>| async move {
+                if let Err(e) = e {
+                    Some(Err(e))
+                } else {
+                    None
+                }
+            });
 
-            let stream = futures::stream::select(fake_stream, rx).boxed();
+            let stream = stream::select(fake_stream, rx).boxed();
 
             Ok(stream)
         })
@@ -222,7 +221,7 @@ impl AsyncMysqlConnection {
     {
         let mut bind_collector = RawBytesBindCollector::<Mysql>::new();
         if let Err(e) = query.collect_binds(&mut bind_collector, &mut (), &Mysql) {
-            return futures::future::ready(Err(e)).boxed();
+            return future::ready(Err(e)).boxed();
         }
 
         let binds = bind_collector.binds;
@@ -250,9 +249,9 @@ impl AsyncMysqlConnection {
         conn: &mut mysql_async::Conn,
         stmt_for_exec: mysql_async::Statement,
         binds: ToSqlHelper,
-        mut tx: futures::channel::mpsc::Sender<QueryResult<MysqlRow>>,
+        mut tx: futures_channel::mpsc::Sender<QueryResult<MysqlRow>>,
     ) -> QueryResult<()> {
-        use futures::SinkExt;
+        use futures_util::sink::SinkExt;
         let res = conn
             .exec_iter(stmt_for_exec, binds)
             .await

src/mysql/mod.rs

@@ -16,10 +16,10 @@ use diesel::pg::{
 use diesel::query_builder::bind_collector::RawBytesBindCollector;
 use diesel::query_builder::{AsQuery, QueryFragment, QueryId};
 use diesel::{ConnectionError, ConnectionResult, QueryResult};
-use futures::future::BoxFuture;
-use futures::lock::Mutex;
-use futures::stream::{BoxStream, TryStreamExt};
-use futures::{Future, FutureExt, StreamExt};
+use futures_util::future::BoxFuture;
+use futures_util::lock::Mutex;
+use futures_util::stream::{BoxStream, TryStreamExt};
+use futures_util::{Future, FutureExt, StreamExt};
 use std::borrow::Cow;
 use std::sync::Arc;
 use tokio_postgres::types::ToSql;
@@ -88,7 +88,7 @@ mod transaction_builder;
 ///       let f2 = q2.get_result::<i32>(conn);
 ///
 ///       // wait on both results
-///       let res = futures::try_join!(f1, f2)?;
+///       let res = futures_util::try_join!(f1, f2)?;
 ///
 ///       assert_eq!(res.0, 1);
 ///       assert_eq!(res.1, 2);
@@ -492,7 +492,7 @@ pub mod tests {
         let f1 = q1.get_result::<i32>(&mut conn);
         let f2 = q2.get_result::<i32>(&mut conn);
 
-        let (r1, r2) = futures::try_join!(f1, f2).unwrap();
+        let (r1, r2) = futures_util::try_join!(f1, f2).unwrap();
 
         assert_eq!(r1, 1);
         assert_eq!(r2, 2);

src/pg/mod.rs

@@ -3,7 +3,7 @@
 //! ```rust
 //! # include!("../doctest_setup.rs");
 //! use diesel::result::Error;
-//! use futures::FutureExt;
+//! use futures_util::FutureExt;
 //! use diesel_async::pooled_connection::AsyncDieselConnectionManager;
 //! use diesel_async::pooled_connection::bb8::Pool;
 //! use diesel_async::RunQueryDsl;

src/pooled_connection/bb8.rs

@@ -3,7 +3,7 @@
 //! ```rust
 //! # include!("../doctest_setup.rs");
 //! use diesel::result::Error;
-//! use futures::FutureExt;
+//! use futures_util::FutureExt;
 //! use diesel_async::pooled_connection::AsyncDieselConnectionManager;
 //! use diesel_async::pooled_connection::deadpool::Pool;
 //! use diesel_async::RunQueryDsl;

src/pooled_connection/deadpool.rs

@@ -3,7 +3,7 @@
 //! ```rust
 //! # include!("../doctest_setup.rs");
 //! use diesel::result::Error;
-//! use futures::FutureExt;
+//! use futures_util::FutureExt;
 //! use diesel_async::pooled_connection::AsyncDieselConnectionManager;
 //! use diesel_async::pooled_connection::mobc::Pool;
 //! use diesel_async::RunQueryDsl;

src/pooled_connection/mobc.rs

@@ -7,7 +7,7 @@
 //! * [mobc](self::mobc)
 use crate::TransactionManager;
 use crate::{AsyncConnection, SimpleAsyncConnection};
-use futures::FutureExt;
+use futures_util::{future, FutureExt};
 use std::fmt;
 use std::ops::DerefMut;
 
@@ -46,8 +46,7 @@ impl std::error::Error for PoolError {}
 /// * [bb8](self::bb8)
 /// * [mobc](self::mobc)
 pub struct AsyncDieselConnectionManager<C> {
-    setup:
-        Box<dyn Fn(&str) -> futures::future::BoxFuture<diesel::ConnectionResult<C>> + Send + Sync>,
+    setup: Box<dyn Fn(&str) -> future::BoxFuture<diesel::ConnectionResult<C>> + Send + Sync>,
     connection_url: String,
 }
 
@@ -68,10 +67,7 @@ impl<C> AsyncDieselConnectionManager<C> {
     /// postgres connection
     pub fn new_with_setup(
         connection_url: impl Into<String>,
-        setup: impl Fn(&str) -> futures::future::BoxFuture<diesel::ConnectionResult<C>>
-            + Send
-            + Sync
-            + 'static,
+        setup: impl Fn(&str) -> future::BoxFuture<diesel::ConnectionResult<C>> + Send + Sync + 'static,
     ) -> Self {
         Self {
             setup: Box::new(setup),

src/pooled_connection/mod.rs

@@ -3,7 +3,7 @@ use diesel::associations::HasTable;
 use diesel::query_builder::IntoUpdateTarget;
 use diesel::result::QueryResult;
 use diesel::AsChangeset;
-use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
+use futures_util::{future, stream, FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
 use std::pin::Pin;
 
 /// The traits used by `QueryDsl`.
@@ -19,7 +19,7 @@ pub mod methods {
     use diesel::expression::QueryMetadata;
     use diesel::query_builder::{AsQuery, QueryFragment, QueryId};
     use diesel::query_dsl::CompatibleType;
-    use futures::{Future, Stream, TryFutureExt};
+    use futures_util::{Future, Stream, TryFutureExt};
 
     /// The `execute` method
     ///
@@ -92,12 +92,12 @@ pub mod methods {
         DB: QueryMetadata<T::SqlType>,
         ST: 'static,
     {
-        type LoadFuture<'conn> = futures::future::MapOk<
+        type LoadFuture<'conn> = future::MapOk<
             Conn::LoadFuture<'conn, 'query>,
             fn(Conn::Stream<'conn, 'query>) -> Self::Stream<'conn>,
         > where Conn: 'conn;
 
-        type Stream<'conn> = futures::stream::Map<
+        type Stream<'conn> = stream::Map<
             Conn::Stream<'conn, 'query>,
             fn(
                 QueryResult<Conn::Row<'conn, 'query>>,
@@ -113,7 +113,7 @@ pub mod methods {
     #[allow(clippy::type_complexity)]
     fn map_result_stream_future<'s, 'a, U, S, R, DB, ST>(
         stream: S,
-    ) -> futures::stream::Map<S, fn(QueryResult<R>) -> QueryResult<U>>
+    ) -> stream::Map<S, fn(QueryResult<R>) -> QueryResult<U>>
     where
         S: Stream<Item = QueryResult<R>> + Send + 's,
         R: diesel::row::Row<'a, DB> + 's,
@@ -144,36 +144,35 @@ pub mod methods {
 pub mod return_futures {
     use super::methods::LoadQuery;
     use diesel::QueryResult;
+    use futures_util::{future, stream};
     use std::pin::Pin;
 
     /// The future returned by [`RunQueryDsl::load`](super::RunQueryDsl::load)
     /// and [`RunQueryDsl::get_results`](super::RunQueryDsl::get_results)
     ///
     /// This is essentially `impl Future<Output = QueryResult<Vec<U>>>`
-    pub type LoadFuture<'conn, 'query, Q: LoadQuery<'query, Conn, U>, Conn, U> =
-        futures::future::AndThen<
-            Q::LoadFuture<'conn>,
-            futures::stream::TryCollect<Q::Stream<'conn>, Vec<U>>,
-            fn(Q::Stream<'conn>) -> futures::stream::TryCollect<Q::Stream<'conn>, Vec<U>>,
-        >;
+    pub type LoadFuture<'conn, 'query, Q: LoadQuery<'query, Conn, U>, Conn, U> = future::AndThen<
+        Q::LoadFuture<'conn>,
+        stream::TryCollect<Q::Stream<'conn>, Vec<U>>,
+        fn(Q::Stream<'conn>) -> stream::TryCollect<Q::Stream<'conn>, Vec<U>>,
+    >;
 
     /// The future returned by [`RunQueryDsl::get_result`](super::RunQueryDsl::get_result)
     ///
     /// This is essentially `impl Future<Output = QueryResult<U>>`
-    pub type GetResult<'conn, 'query, Q: LoadQuery<'query, Conn, U>, Conn, U> =
-        futures::future::AndThen<
-            Q::LoadFuture<'conn>,
-            futures::future::Map<
-                futures::stream::StreamFuture<Pin<Box<Q::Stream<'conn>>>>,
-                fn((Option<QueryResult<U>>, Pin<Box<Q::Stream<'conn>>>)) -> QueryResult<U>,
-            >,
-            fn(
-                Q::Stream<'conn>,
-            ) -> futures::future::Map<
-                futures::stream::StreamFuture<Pin<Box<Q::Stream<'conn>>>>,
-                fn((Option<QueryResult<U>>, Pin<Box<Q::Stream<'conn>>>)) -> QueryResult<U>,
-            >,
-        >;
+    pub type GetResult<'conn, 'query, Q: LoadQuery<'query, Conn, U>, Conn, U> = future::AndThen<
+        Q::LoadFuture<'conn>,
+        future::Map<
+            stream::StreamFuture<Pin<Box<Q::Stream<'conn>>>>,
+            fn((Option<QueryResult<U>>, Pin<Box<Q::Stream<'conn>>>)) -> QueryResult<U>,
+        >,
+        fn(
+            Q::Stream<'conn>,
+        ) -> future::Map<
+            stream::StreamFuture<Pin<Box<Q::Stream<'conn>>>>,
+            fn((Option<QueryResult<U>>, Pin<Box<Q::Stream<'conn>>>)) -> QueryResult<U>,
+        >,
+    >;
 }
 
 /// Methods used to execute queries.
@@ -329,7 +328,7 @@ pub trait RunQueryDsl<Conn>: Sized {
         Conn: AsyncConnection,
         Self: methods::LoadQuery<'query, Conn, U> + 'query,
     {
-        fn collect_result<U, S>(stream: S) -> futures::stream::TryCollect<S, Vec<U>>
+        fn collect_result<U, S>(stream: S) -> stream::TryCollect<S, Vec<U>>
         where
             S: Stream<Item = QueryResult<U>>,
         {
@@ -373,14 +372,14 @@ pub trait RunQueryDsl<Conn>: Sized {
     /// # async fn run_test() -> QueryResult<()> {
     /// #     use diesel::insert_into;
     /// #     use schema::users::dsl::*;
-    /// #     use futures::stream::TryStreamExt;
+    /// #     use futures_util::stream::TryStreamExt;
     /// #     let connection = &mut establish_connection().await;
     /// let data = users.select(name)
     ///     .load_stream::<String>(connection)
     ///     .await?
     ///     .try_fold(Vec::new(), |mut acc, item| {
     ///          acc.push(item);
-    ///          futures::future::ready(Ok(acc))
+    ///          futures_util::future::ready(Ok(acc))
     ///      })
     ///     .await?;
     /// assert_eq!(vec!["Sean", "Tess"], data);
@@ -401,14 +400,14 @@ pub trait RunQueryDsl<Conn>: Sized {
     /// # async fn run_test() -> QueryResult<()> {
     /// #     use diesel::insert_into;
     /// #     use schema::users::dsl::*;
-    /// #     use futures::stream::TryStreamExt;
+    /// #     use futures_util::stream::TryStreamExt;
     /// #     let connection = &mut establish_connection().await;
     /// let data = users
     ///     .load_stream::<(i32, String)>(connection)
     ///     .await?
     ///     .try_fold(Vec::new(), |mut acc, item| {
     ///          acc.push(item);
-    ///          futures::future::ready(Ok(acc))
+    ///          futures_util::future::ready(Ok(acc))
     ///      })
     ///     .await?;
     /// let expected_data = vec![
@@ -439,14 +438,14 @@ pub trait RunQueryDsl<Conn>: Sized {
     /// # async fn run_test() -> QueryResult<()> {
     /// #     use diesel::insert_into;
     /// #     use schema::users::dsl::*;
-    /// #     use futures::stream::TryStreamExt;
+    /// #     use futures_util::stream::TryStreamExt;
     /// #     let connection = &mut establish_connection().await;
     /// let data = users
     ///     .load_stream::<User>(connection)
     ///     .await?
     ///     .try_fold(Vec::new(), |mut acc, item| {
     ///          acc.push(item);
-    ///          futures::future::ready(Ok(acc))
+    ///          futures_util::future::ready(Ok(acc))
     ///      })
     ///     .await?;
     /// let expected_data = vec![
@@ -526,8 +525,8 @@ pub trait RunQueryDsl<Conn>: Sized {
         #[allow(clippy::type_complexity)]
         fn get_next_stream_element<S, U>(
             stream: S,
-        ) -> futures::future::Map<
-            futures::stream::StreamFuture<Pin<Box<S>>>,
+        ) -> future::Map<
+            stream::StreamFuture<Pin<Box<S>>>,
             fn((Option<QueryResult<U>>, Pin<Box<S>>)) -> QueryResult<U>,
         >
         where