Merge pull request #11 from weiznich/pipelining_support

Support for connection pooling via deadpool, bb8 and mobc

Author: weiznich

Cargo.lock

@@ -17,18 +17,21 @@ tokio = { version = "1.12.0", features = ["rt"], optional = true}
 mysql_async = { version = "0.29.0", optional = true}
 mysql_common = {version = "0.28.0", optional = true}
 
+bb8 = {version = "0.8", optional = true}
+deadpool = {version = "0.9", optional = true}
+mobc = {version = "0.7", optional = true}
+
 [dev-dependencies]
 tokio = {version = "1.12.0", features = ["rt", "macros"]}
 cfg-if = "1"
 chrono = "0.4"
 diesel = { version = "2.0.0-rc.0", default-features = false,  features = ["chrono"]}
 
 [features]
-default = ["postgres", "mysql"]
+default = ["postgres", "mysql", "deadpool", "bb8", "mobc"]
 mysql = ["diesel/mysql_backend", "mysql_async", "mysql_common"]
 postgres = ["diesel/postgres_backend", "tokio-postgres", "tokio", "tokio/rt-multi-thread"]
 
-
 [[test]]
 name = "integration_tests"
 path = "tests/lib.rs"

Cargo.toml

@@ -75,6 +75,8 @@ mod pg;
 mod run_query_dsl;
 mod stmt_cache;
 mod transaction_manager;
+#[cfg(any(feature = "deadpool", feature = "bb8", feature = "mobc"))]
+pub mod pooled_connection;
 
 #[cfg(feature = "mysql")]
 pub use self::mysql::AsyncMysqlConnection;

src/lib.rs

@@ -213,3 +213,6 @@ impl AsyncMysqlConnection {
         }).boxed()
     }
 }
+
+#[cfg(any(feature = "deadpool", feature = "bb8", feature = "mobc"))]
+impl crate::pooled_connection::PoolableConnection for AsyncMysqlConnection {}

src/mysql/mod.rs

@@ -482,26 +482,8 @@ async fn lookup_type(
     Ok((r.get(0), r.get(1)))
 }
 
-diesel::table! {
-    pg_type (oid) {
-        oid -> Oid,
-        typname -> Text,
-        typarray -> Oid,
-        typnamespace -> Oid,
-    }
-}
-
-diesel::table! {
-    pg_namespace (oid) {
-        oid -> Oid,
-        nspname -> Text,
-    }
-}
-
-diesel::joinable!(pg_type -> pg_namespace(typnamespace));
-diesel::allow_tables_to_appear_in_same_query!(pg_type, pg_namespace);
-
-diesel::sql_function! { fn pg_my_temp_schema() -> Oid; }
+#[cfg(any(feature = "deadpool", feature = "bb8", feature = "mobc"))]
+impl crate::pooled_connection::PoolableConnection for AsyncPgConnection {}
 
 #[cfg(test)]
 pub mod tests {

src/pg/mod.rs

@@ -0,0 +1,67 @@
+//! A pool implementation for `diesel-async` based on [`bb8`]
+//!
+//! ```rust
+//! # include!("../doctest_setup.rs");
+//! use diesel::result::Error;
+//! use futures::FutureExt;
+//! use diesel_async::pooled_connection::AsyncDieselConnectionManager;
+//! use diesel_async::pooled_connection::bb8::Pool;
+//!
+//! # #[tokio::main(flavor = "current_thread")]
+//! # async fn main() {
+//! #     run_test().await.unwrap();
+//! # }
+//! #
+//! # #[cfg(feature = "postgres")]
+//! # fn get_config() -> AsyncDieselConnectionManager<diesel_async::AsyncPgConnection> {
+//! #     let db_url = database_url_from_env("PG_DATABASE_URL");
+//!   let config = AsyncDieselConnectionManager::<diesel_async::AsyncPgConnection>::new(db_url);
+//! #     config
+//! #  }
+//! #
+//! # #[cfg(feature = "mysql")]
+//! # fn get_config() -> AsyncDieselConnectionManager<diesel_async::AsyncMysqlConnection> {
+//! #     let db_url = database_url_from_env("MYSQL_DATABASE_URL");
+//! #    let config = AsyncDieselConnectionManager::<diesel_async::AsyncMysqlConnection>::new(db_url);
+//! #     config
+//! #  }
+//! #
+//! # async fn run_test() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
+//! #     use schema::users::dsl::*;
+//! #     let config = get_config();
+//!   let pool = Pool::builder().build(config).await?;
+//!   let mut conn = pool.get().await?;
+//!   let res = users.load::<(i32, String)>(&mut conn).await?;
+//! #     Ok(())
+//! # }
+//! ```
+
+use super::{AsyncDieselConnectionManager, PoolError, PoolableConnection};
+use bb8::ManageConnection;
+
+/// Type alias for using [`bb8::Pool`] with [`diesel-async`]
+pub type Pool<C> = bb8::Pool<AsyncDieselConnectionManager<C>>;
+
+#[async_trait::async_trait]
+impl<C> ManageConnection for AsyncDieselConnectionManager<C>
+where
+    C: PoolableConnection + 'static,
+{
+    type Connection = C;
+
+    type Error = PoolError;
+
+    async fn connect(&self) -> Result<Self::Connection, Self::Error> {
+        C::establish(&self.connection_url)
+            .await
+            .map_err(PoolError::ConnectionError)
+    }
+
+    async fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+        conn.ping().await.map_err(PoolError::QueryError)
+    }
+
+    fn has_broken(&self, conn: &mut Self::Connection) -> bool {
+        std::thread::panicking() || conn.is_broken()
+    }
+}