From 2f3bb68a12ab7631a3bd99b1c74feb6af750c89a Mon Sep 17 00:00:00 2001 From: Paolo Barbolini Date: Sun, 1 Jun 2025 17:38:29 +0200 Subject: [PATCH 1/5] Uses std APIs directly instead of `futures-util` re-exports --- src/async_connection_wrapper.rs | 2 +- src/lib.rs | 3 ++- src/mysql/mod.rs | 3 ++- src/pg/mod.rs | 3 ++- 4 files changed, 7 insertions(+), 4 deletions(-) diff --git a/src/async_connection_wrapper.rs b/src/async_connection_wrapper.rs index 3a709cb..4e1c0b3 100644 --- a/src/async_connection_wrapper.rs +++ b/src/async_connection_wrapper.rs @@ -9,9 +9,9 @@ //! as replacement for the existing connection //! implementations provided by diesel -use futures_util::Future; use futures_util::Stream; use futures_util::StreamExt; +use std::future::Future; use std::pin::Pin; /// This is a helper trait that allows to customize the diff --git a/src/lib.rs b/src/lib.rs index 5ae0136..7a0cf4d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -79,8 +79,9 @@ use diesel::query_builder::{AsQuery, QueryFragment, QueryId}; use diesel::row::Row; use diesel::{ConnectionResult, QueryResult}; use futures_util::future::BoxFuture; -use futures_util::{Future, FutureExt, Stream}; +use futures_util::{FutureExt, Stream}; use std::fmt::Debug; +use std::future::Future; pub use scoped_futures; use scoped_futures::{ScopedBoxFuture, ScopedFutureExt}; diff --git a/src/mysql/mod.rs b/src/mysql/mod.rs index 6f2321f..49100c1 100644 --- a/src/mysql/mod.rs +++ b/src/mysql/mod.rs @@ -13,9 +13,10 @@ use diesel::result::{ConnectionError, ConnectionResult}; use diesel::QueryResult; use futures_util::future::BoxFuture; use futures_util::stream::{self, BoxStream}; -use futures_util::{Future, FutureExt, StreamExt, TryStreamExt}; +use futures_util::{FutureExt, StreamExt, TryStreamExt}; use mysql_async::prelude::Queryable; use mysql_async::{Opts, OptsBuilder, Statement}; +use std::future::Future; mod error_helper; mod row; diff --git a/src/pg/mod.rs b/src/pg/mod.rs index ce24ba8..181b500 100644 --- a/src/pg/mod.rs +++ b/src/pg/mod.rs @@ -26,8 +26,9 @@ use futures_util::future::BoxFuture; use futures_util::future::Either; use futures_util::stream::{BoxStream, TryStreamExt}; use futures_util::TryFutureExt; -use futures_util::{Future, FutureExt, StreamExt}; +use futures_util::{FutureExt, StreamExt}; use std::collections::{HashMap, HashSet}; +use std::future::Future; use std::sync::Arc; use tokio::sync::broadcast; use tokio::sync::oneshot; From 76953bc8375703623e5b8ab91622de40010f18bd Mon Sep 17 00:00:00 2001 From: Paolo Barbolini Date: Sun, 1 Jun 2025 17:42:01 +0200 Subject: [PATCH 2/5] Uses `futures-core` APIs directly instead of `futures-util` re-exports --- Cargo.toml | 1 + src/async_connection_wrapper.rs | 2 +- src/lib.rs | 5 +++-- src/mysql/mod.rs | 5 +++-- src/pg/mod.rs | 5 +++-- src/pooled_connection/mod.rs | 9 ++++----- src/run_query_dsl/mod.rs | 5 +++-- src/stmt_cache.rs | 3 ++- src/sync_connection_wrapper/mod.rs | 4 ++-- 9 files changed, 22 insertions(+), 17 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 291ea44..9148b65 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ description = "An async extension for Diesel the safe, extensible ORM and Query rust-version = "1.82.0" [dependencies] +futures-core = "0.3.17" futures-channel = { version = "0.3.17", default-features = false, features = [ "std", "sink", diff --git a/src/async_connection_wrapper.rs b/src/async_connection_wrapper.rs index 4e1c0b3..62ce265 100644 --- a/src/async_connection_wrapper.rs +++ b/src/async_connection_wrapper.rs @@ -9,7 +9,7 @@ //! as replacement for the existing connection //! implementations provided by diesel -use futures_util::Stream; +use futures_core::Stream; use futures_util::StreamExt; use std::future::Future; use std::pin::Pin; diff --git a/src/lib.rs b/src/lib.rs index 7a0cf4d..e84448f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -78,8 +78,9 @@ use diesel::connection::{CacheSize, Instrumentation}; use diesel::query_builder::{AsQuery, QueryFragment, QueryId}; use diesel::row::Row; use diesel::{ConnectionResult, QueryResult}; -use futures_util::future::BoxFuture; -use futures_util::{FutureExt, Stream}; +use futures_core::future::BoxFuture; +use futures_core::Stream; +use futures_util::FutureExt; use std::fmt::Debug; use std::future::Future; diff --git a/src/mysql/mod.rs b/src/mysql/mod.rs index 49100c1..b25e5e0 100644 --- a/src/mysql/mod.rs +++ b/src/mysql/mod.rs @@ -11,8 +11,9 @@ use diesel::query_builder::QueryBuilder; use diesel::query_builder::{bind_collector::RawBytesBindCollector, QueryFragment, QueryId}; use diesel::result::{ConnectionError, ConnectionResult}; use diesel::QueryResult; -use futures_util::future::BoxFuture; -use futures_util::stream::{self, BoxStream}; +use futures_core::future::BoxFuture; +use futures_core::stream::BoxStream; +use futures_util::stream; use futures_util::{FutureExt, StreamExt, TryStreamExt}; use mysql_async::prelude::Queryable; use mysql_async::{Opts, OptsBuilder, Statement}; diff --git a/src/pg/mod.rs b/src/pg/mod.rs index 181b500..9a8450c 100644 --- a/src/pg/mod.rs +++ b/src/pg/mod.rs @@ -22,9 +22,10 @@ use diesel::query_builder::bind_collector::RawBytesBindCollector; use diesel::query_builder::{AsQuery, QueryBuilder, QueryFragment, QueryId}; use diesel::result::{DatabaseErrorKind, Error}; use diesel::{ConnectionError, ConnectionResult, QueryResult}; -use futures_util::future::BoxFuture; +use futures_core::future::BoxFuture; +use futures_core::stream::BoxStream; use futures_util::future::Either; -use futures_util::stream::{BoxStream, TryStreamExt}; +use futures_util::stream::TryStreamExt; use futures_util::TryFutureExt; use futures_util::{FutureExt, StreamExt}; use std::collections::{HashMap, HashSet}; diff --git a/src/pooled_connection/mod.rs b/src/pooled_connection/mod.rs index 4674d22..f155d3c 100644 --- a/src/pooled_connection/mod.rs +++ b/src/pooled_connection/mod.rs @@ -10,8 +10,8 @@ use crate::{TransactionManager, UpdateAndFetchResults}; use diesel::associations::HasTable; use diesel::connection::{CacheSize, Instrumentation}; use diesel::QueryResult; -use futures_util::future::BoxFuture; -use futures_util::{future, FutureExt}; +use futures_core::future::BoxFuture; +use futures_util::FutureExt; use std::borrow::Cow; use std::fmt; use std::future::Future; @@ -47,11 +47,10 @@ impl std::error::Error for PoolError {} /// Type of the custom setup closure passed to [`ManagerConfig::custom_setup`] pub type SetupCallback = - Box future::BoxFuture> + Send + Sync>; + Box BoxFuture> + Send + Sync>; /// Type of the recycle check callback for the [`RecyclingMethod::CustomFunction`] variant -pub type RecycleCheckCallback = - dyn Fn(&mut C) -> future::BoxFuture> + Send + Sync; +pub type RecycleCheckCallback = dyn Fn(&mut C) -> BoxFuture> + Send + Sync; /// Possible methods of how a connection is recycled. #[derive(Default)] diff --git a/src/run_query_dsl/mod.rs b/src/run_query_dsl/mod.rs index 0ee56a7..6d98cff 100644 --- a/src/run_query_dsl/mod.rs +++ b/src/run_query_dsl/mod.rs @@ -3,8 +3,9 @@ use diesel::associations::HasTable; use diesel::query_builder::IntoUpdateTarget; use diesel::result::QueryResult; use diesel::AsChangeset; -use futures_util::future::BoxFuture; -use futures_util::{future, stream, FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt}; +use futures_core::future::BoxFuture; +use futures_core::Stream; +use futures_util::{future, stream, FutureExt, StreamExt, TryFutureExt, TryStreamExt}; use std::future::Future; use std::pin::Pin; diff --git a/src/stmt_cache.rs b/src/stmt_cache.rs index cd3ccc5..b8e5a66 100644 --- a/src/stmt_cache.rs +++ b/src/stmt_cache.rs @@ -1,5 +1,6 @@ use diesel::connection::statement_cache::{MaybeCached, StatementCallbackReturnType}; use diesel::QueryResult; +use futures_core::future::BoxFuture; use futures_util::{future, FutureExt, TryFutureExt}; use std::future::Future; @@ -7,7 +8,7 @@ pub(crate) struct CallbackHelper(pub(crate) F); type PrepareFuture<'a, C, S> = future::Either< future::Ready, C)>>, - future::BoxFuture<'a, QueryResult<(MaybeCached<'a, S>, C)>>, + BoxFuture<'a, QueryResult<(MaybeCached<'a, S>, C)>>, >; impl StatementCallbackReturnType for CallbackHelper diff --git a/src/sync_connection_wrapper/mod.rs b/src/sync_connection_wrapper/mod.rs index 75a2122..2bbf570 100644 --- a/src/sync_connection_wrapper/mod.rs +++ b/src/sync_connection_wrapper/mod.rs @@ -6,7 +6,7 @@ //! //! * using a sync Connection implementation in async context //! * using the same code base for async crates needing multiple backends -use futures_util::future::BoxFuture; +use futures_core::future::BoxFuture; use std::error::Error; #[cfg(feature = "sqlite")] @@ -100,7 +100,7 @@ mod implementation { }; use diesel::row::IntoOwnedRow; use diesel::{ConnectionResult, QueryResult}; - use futures_util::stream::BoxStream; + use futures_core::stream::BoxStream; use futures_util::{FutureExt, StreamExt, TryFutureExt}; use std::marker::PhantomData; use std::sync::{Arc, Mutex}; From 13b89602ae6ecdc9c4ffefb9dca6d26707c2251c Mon Sep 17 00:00:00 2001 From: Paolo Barbolini Date: Sun, 1 Jun 2025 17:48:46 +0200 Subject: [PATCH 3/5] Replace stabilized `futures-util` APIs with std --- src/lib.rs | 4 ++-- src/run_query_dsl/mod.rs | 6 +++--- src/stmt_cache.rs | 15 ++++++++------- 3 files changed, 13 insertions(+), 12 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index e84448f..97929bd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -324,7 +324,7 @@ pub trait AsyncConnection: SimpleAsyncConnection + Sized + Send { .map_err(|_| diesel::result::Error::RollbackTransaction) .and_then(move |r| { let _ = user_result_tx.send(r); - futures_util::future::ready(Err(diesel::result::Error::RollbackTransaction)) + std::future::ready(Err(diesel::result::Error::RollbackTransaction)) }) .scope_boxed() }) @@ -332,7 +332,7 @@ pub trait AsyncConnection: SimpleAsyncConnection + Sized + Send { let r = user_result_rx .try_recv() .expect("Transaction did not succeed"); - futures_util::future::ready(r) + std::future::ready(r) }) } diff --git a/src/run_query_dsl/mod.rs b/src/run_query_dsl/mod.rs index 6d98cff..b1ed693 100644 --- a/src/run_query_dsl/mod.rs +++ b/src/run_query_dsl/mod.rs @@ -399,7 +399,7 @@ pub trait RunQueryDsl: Sized { /// .await? /// .try_fold(Vec::new(), |mut acc, item| { /// acc.push(item); - /// futures_util::future::ready(Ok(acc)) + /// std::future::ready(Ok(acc)) /// }) /// .await?; /// assert_eq!(vec!["Sean", "Tess"], data); @@ -428,7 +428,7 @@ pub trait RunQueryDsl: Sized { /// .await? /// .try_fold(Vec::new(), |mut acc, item| { /// acc.push(item); - /// futures_util::future::ready(Ok(acc)) + /// std::future::ready(Ok(acc)) /// }) /// .await?; /// let expected_data = vec![ @@ -468,7 +468,7 @@ pub trait RunQueryDsl: Sized { /// .await? /// .try_fold(Vec::new(), |mut acc, item| { /// acc.push(item); - /// futures_util::future::ready(Ok(acc)) + /// std::future::ready(Ok(acc)) /// }) /// .await?; /// let expected_data = vec![ diff --git a/src/stmt_cache.rs b/src/stmt_cache.rs index b8e5a66..c2270b8 100644 --- a/src/stmt_cache.rs +++ b/src/stmt_cache.rs @@ -1,12 +1,13 @@ use diesel::connection::statement_cache::{MaybeCached, StatementCallbackReturnType}; use diesel::QueryResult; use futures_core::future::BoxFuture; -use futures_util::{future, FutureExt, TryFutureExt}; -use std::future::Future; +use futures_util::future::Either; +use futures_util::{FutureExt, TryFutureExt}; +use std::future::{self, Future}; pub(crate) struct CallbackHelper(pub(crate) F); -type PrepareFuture<'a, C, S> = future::Either< +type PrepareFuture<'a, C, S> = Either< future::Ready, C)>>, BoxFuture<'a, QueryResult<(MaybeCached<'a, S>, C)>>, >; @@ -19,14 +20,14 @@ where type Return<'a> = PrepareFuture<'a, C, S>; fn from_error<'a>(e: diesel::result::Error) -> Self::Return<'a> { - future::Either::Left(future::ready(Err(e))) + Either::Left(future::ready(Err(e))) } fn map_to_no_cache<'a>(self) -> Self::Return<'a> where Self: 'a, { - future::Either::Right( + Either::Right( self.0 .map_ok(|(stmt, conn)| (MaybeCached::CannotCache(stmt), conn)) .boxed(), @@ -34,7 +35,7 @@ where } fn map_to_cache(stmt: &mut S, conn: C) -> Self::Return<'_> { - future::Either::Left(future::ready(Ok((MaybeCached::Cached(stmt), conn)))) + Either::Left(future::ready(Ok((MaybeCached::Cached(stmt), conn)))) } fn register_cache<'a>( @@ -44,7 +45,7 @@ where where Self: 'a, { - future::Either::Right( + Either::Right( self.0 .map_ok(|(stmt, conn)| (MaybeCached::Cached(callback(stmt)), conn)) .boxed(), From ee0223d66512a0e4973ea82a7ffb3083f550f139 Mon Sep 17 00:00:00 2001 From: Paolo Barbolini Date: Sun, 1 Jun 2025 18:18:20 +0200 Subject: [PATCH 4/5] Replace `futures_util::try_join!` with `futures_util::future::try_join` --- src/pg/mod.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/pg/mod.rs b/src/pg/mod.rs index 9a8450c..2aa045b 100644 --- a/src/pg/mod.rs +++ b/src/pg/mod.rs @@ -445,10 +445,11 @@ impl AsyncPgConnection { async fn set_config_options(&mut self) -> QueryResult<()> { use crate::run_query_dsl::RunQueryDsl; - futures_util::try_join!( + futures_util::future::try_join( diesel::sql_query("SET TIME ZONE 'UTC'").execute(self), diesel::sql_query("SET CLIENT_ENCODING TO 'UTF8'").execute(self), - )?; + ) + .await?; Ok(()) } From dd64b6e591b9cfd64ca09420cefbc12c0268a2d0 Mon Sep 17 00:00:00 2001 From: Paolo Barbolini Date: Sun, 1 Jun 2025 17:55:28 +0200 Subject: [PATCH 5/5] Reduce `futures-util` features --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 9148b65..bae1ecb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,7 @@ futures-channel = { version = "0.3.17", default-features = false, features = [ "sink", ], optional = true } futures-util = { version = "0.3.17", default-features = false, features = [ - "std", + "alloc", "sink", ] } tokio-postgres = { version = "0.7.10", optional = true }