Copy new statement cache form diesel to unblock a release

Author: weiznich

Cargo.toml

@@ -43,8 +43,6 @@ default-features = false
 features = [
   "i-implement-a-third-party-backend-and-opt-into-breaking-changes",
 ]
-git = "https://github.com/diesel-rs/diesel"
-branch = "master"
 
 [dev-dependencies]
 tokio = { version = "1.12.0", features = ["rt", "macros", "rt-multi-thread"] }
@@ -58,13 +56,9 @@ default-features = false
 features = [
   "chrono"
 ]
-git = "https://github.com/diesel-rs/diesel"
-branch = "master"
 
 [dev-dependencies.diesel_migrations]
 version = "2.2.0"
-git = "https://github.com/diesel-rs/diesel"
-branch = "master"
 
 [features]
 default = []

src/async_connection_wrapper.rs

@@ -100,7 +100,7 @@ pub type AsyncConnectionWrapper<C, B = self::implementation::Tokio> =
 pub use self::implementation::AsyncConnectionWrapper;
 
 mod implementation {
-    use diesel::connection::{CacheSize, Instrumentation, SimpleConnection};
+    use diesel::connection::{Instrumentation, SimpleConnection};
     use std::ops::{Deref, DerefMut};
 
     use super::*;
@@ -187,10 +187,6 @@ mod implementation {
         fn set_instrumentation(&mut self, instrumentation: impl Instrumentation) {
             self.inner.set_instrumentation(instrumentation);
         }
-
-        fn set_prepared_statement_cache_size(&mut self, size: CacheSize) {
-            self.inner.set_prepared_statement_cache_size(size)
-        }
     }
 
     impl<C, B> diesel::connection::LoadConnection for AsyncConnectionWrapper<C, B>

src/lib.rs

@@ -72,9 +72,8 @@
     clippy::cast_possible_truncation,
     clippy::cast_sign_loss
 )]
-
 use diesel::backend::Backend;
-use diesel::connection::{CacheSize, Instrumentation};
+use diesel::connection::Instrumentation;
 use diesel::query_builder::{AsQuery, QueryFragment, QueryId};
 use diesel::row::Row;
 use diesel::{ConnectionResult, QueryResult};
@@ -96,6 +95,7 @@ pub mod pg;
 #[cfg(feature = "pool")]
 pub mod pooled_connection;
 mod run_query_dsl;
+mod statement_cache;
 #[cfg(any(feature = "postgres", feature = "mysql"))]
 mod stmt_cache;
 #[cfg(feature = "sync-connection-wrapper")]
@@ -111,6 +111,8 @@ pub use self::pg::AsyncPgConnection;
 #[doc(inline)]
 pub use self::run_query_dsl::*;
 
+#[doc(inline)]
+pub use self::statement_cache::CacheSize;
 #[doc(inline)]
 pub use self::transaction_manager::{AnsiTransactionManager, TransactionManager};
 

src/mysql/mod.rs

@@ -1,11 +1,10 @@
+use crate::statement_cache::CacheSize;
+use crate::statement_cache::{MaybeCached, QueryFragmentForCachedStatement, StatementCache};
 use crate::stmt_cache::{CallbackHelper, QueryFragmentHelper};
 use crate::{AnsiTransactionManager, AsyncConnection, SimpleAsyncConnection};
-use diesel::connection::statement_cache::{
-    MaybeCached, QueryFragmentForCachedStatement, StatementCache,
-};
+use diesel::connection::Instrumentation;
+use diesel::connection::InstrumentationEvent;
 use diesel::connection::StrQueryHelper;
-use diesel::connection::{CacheSize, Instrumentation};
-use diesel::connection::{DynInstrumentation, InstrumentationEvent};
 use diesel::mysql::{Mysql, MysqlQueryBuilder, MysqlType};
 use diesel::query_builder::QueryBuilder;
 use diesel::query_builder::{bind_collector::RawBytesBindCollector, QueryFragment, QueryId};
@@ -33,7 +32,7 @@ pub struct AsyncMysqlConnection {
     conn: mysql_async::Conn,
     stmt_cache: StatementCache<Mysql, Statement>,
     transaction_manager: AnsiTransactionManager,
-    instrumentation: DynInstrumentation,
+    instrumentation: Option<Box<dyn Instrumentation>>,
 }
 
 impl SimpleAsyncConnection for AsyncMysqlConnection {
@@ -74,7 +73,7 @@ impl AsyncConnection for AsyncMysqlConnection {
     type TransactionManager = AnsiTransactionManager;
 
     async fn establish(database_url: &str) -> diesel::ConnectionResult<Self> {
-        let mut instrumentation = DynInstrumentation::default_instrumentation();
+        let mut instrumentation = diesel::connection::get_default_instrumentation();
         instrumentation.on_connection_event(InstrumentationEvent::start_establish_connection(
             database_url,
         ));
@@ -99,10 +98,6 @@ impl AsyncConnection for AsyncMysqlConnection {
             let stmt_for_exec = match stmt {
                 MaybeCached::Cached(ref s) => (*s).clone(),
                 MaybeCached::CannotCache(ref s) => s.clone(),
-                _ => unreachable!(
-                    "Diesel has only two variants here at the time of writing.\n\
-                     If you ever see this error message please open in issue in the diesel-async issue tracker"
-                ),
             };
 
             let (tx, rx) = futures_channel::mpsc::channel(0);
@@ -179,11 +174,11 @@ impl AsyncConnection for AsyncMysqlConnection {
     }
 
     fn instrumentation(&mut self) -> &mut dyn Instrumentation {
-        &mut *self.instrumentation
+        &mut self.instrumentation
     }
 
     fn set_instrumentation(&mut self, instrumentation: impl Instrumentation) {
-        self.instrumentation = instrumentation.into();
+        self.instrumentation = Some(Box::new(instrumentation));
     }
 
     fn set_prepared_statement_cache_size(&mut self, size: CacheSize) {
@@ -211,7 +206,7 @@ fn update_transaction_manager_status<T>(
 fn prepare_statement_helper<'a>(
     conn: &'a mut mysql_async::Conn,
     sql: &str,
-    _is_for_cache: diesel::connection::statement_cache::PrepareForCache,
+    _is_for_cache: crate::statement_cache::PrepareForCache,
     _metadata: &[MysqlType],
 ) -> CallbackHelper<impl Future<Output = QueryResult<(Statement, &'a mut mysql_async::Conn)>> + Send>
 {
@@ -239,7 +234,7 @@ impl AsyncMysqlConnection {
             conn,
             stmt_cache: StatementCache::new(),
             transaction_manager: AnsiTransactionManager::default(),
-            instrumentation: DynInstrumentation::default_instrumentation(),
+            instrumentation: diesel::connection::get_default_instrumentation(),
         };
 
         for stmt in CONNECTION_SETUP_QUERIES {
@@ -296,21 +291,25 @@ impl AsyncMysqlConnection {
                 sql,
                 safe_to_cache: is_safe_to_cache_prepared,
             };
+
             let inner = async {
-                let (stmt, conn) = stmt_cache
-                    .cached_statement_non_generic(
-                        query_id,
-                        &helper,
-                        &Mysql,
-                        &metadata,
-                        conn,
-                        prepare_statement_helper,
-                        &mut **instrumentation,
-                    )
-                    .await?;
+                let (stmt, conn) = {
+                    stmt_cache
+                        .cached_statement_non_generic(
+                            query_id,
+                            &helper,
+                            &Mysql,
+                            &metadata,
+                            conn,
+                            prepare_statement_helper,
+                            &mut *instrumentation,
+                        )
+                        .await?
+                };
                 callback(conn, stmt, ToSqlHelper { metadata, binds }).await
             };
             let r = update_transaction_manager_status(inner.await, transaction_manager);
+
             instrumentation.on_connection_event(InstrumentationEvent::finish_query(
                 &StrQueryHelper::new(&helper.sql),
                 r.as_ref().err(),
@@ -371,7 +370,7 @@ impl AsyncMysqlConnection {
             conn,
             stmt_cache: StatementCache::new(),
             transaction_manager: AnsiTransactionManager::default(),
-            instrumentation: DynInstrumentation::none(),
+            instrumentation: diesel::connection::get_default_instrumentation(),
         })
     }
 }

src/pg/mod.rs

@@ -7,14 +7,13 @@
 use self::error_helper::ErrorHelper;
 use self::row::PgRow;
 use self::serialize::ToSqlHelper;
+use crate::statement_cache::CacheSize;
+use crate::statement_cache::{PrepareForCache, QueryFragmentForCachedStatement, StatementCache};
 use crate::stmt_cache::{CallbackHelper, QueryFragmentHelper};
 use crate::{AnsiTransactionManager, AsyncConnection, SimpleAsyncConnection};
-use diesel::connection::statement_cache::{
-    PrepareForCache, QueryFragmentForCachedStatement, StatementCache,
-};
+use diesel::connection::Instrumentation;
+use diesel::connection::InstrumentationEvent;
 use diesel::connection::StrQueryHelper;
-use diesel::connection::{CacheSize, Instrumentation};
-use diesel::connection::{DynInstrumentation, InstrumentationEvent};
 use diesel::pg::{
     Pg, PgMetadataCache, PgMetadataCacheKey, PgMetadataLookup, PgQueryBuilder, PgTypeMetadata,
 };
@@ -132,7 +131,7 @@ pub struct AsyncPgConnection {
     connection_future: Option<broadcast::Receiver<Arc<tokio_postgres::Error>>>,
     shutdown_channel: Option<oneshot::Sender<()>>,
     // a sync mutex is fine here as we only hold it for a really short time
-    instrumentation: Arc<std::sync::Mutex<DynInstrumentation>>,
+    instrumentation: Arc<std::sync::Mutex<Option<Box<dyn Instrumentation>>>>,
 }
 
 impl SimpleAsyncConnection for AsyncPgConnection {
@@ -169,7 +168,7 @@ impl AsyncConnection for AsyncPgConnection {
     type TransactionManager = AnsiTransactionManager;
 
     async fn establish(database_url: &str) -> ConnectionResult<Self> {
-        let mut instrumentation = DynInstrumentation::default_instrumentation();
+        let mut instrumentation = diesel::connection::get_default_instrumentation();
         instrumentation.on_connection_event(InstrumentationEvent::start_establish_connection(
             database_url,
         ));
@@ -236,14 +235,14 @@ impl AsyncConnection for AsyncPgConnection {
         // that means there is only one instance of this arc and
         // we can simply access the inner data
         if let Some(instrumentation) = Arc::get_mut(&mut self.instrumentation) {
-            &mut **(instrumentation.get_mut().unwrap_or_else(|p| p.into_inner()))
+            &mut *(instrumentation.get_mut().unwrap_or_else(|p| p.into_inner()))
         } else {
             panic!("Cannot access shared instrumentation")
         }
     }
 
     fn set_instrumentation(&mut self, instrumentation: impl Instrumentation) {
-        self.instrumentation = Arc::new(std::sync::Mutex::new(instrumentation.into()));
+        self.instrumentation = Arc::new(std::sync::Mutex::new(Some(Box::new(instrumentation))));
     }
 
     fn set_prepared_statement_cache_size(&mut self, size: CacheSize) {
@@ -395,7 +394,7 @@ impl AsyncPgConnection {
             None,
             None,
             Arc::new(std::sync::Mutex::new(
-                DynInstrumentation::default_instrumentation(),
+                diesel::connection::get_default_instrumentation(),
             )),
         )
         .await
@@ -416,7 +415,9 @@ impl AsyncPgConnection {
             client,
             Some(error_rx),
             Some(shutdown_tx),
-            Arc::new(std::sync::Mutex::new(DynInstrumentation::none())),
+            Arc::new(std::sync::Mutex::new(
+                diesel::connection::get_default_instrumentation(),
+            )),
         )
         .await
     }
@@ -425,7 +426,7 @@ impl AsyncPgConnection {
         conn: tokio_postgres::Client,
         connection_future: Option<broadcast::Receiver<Arc<tokio_postgres::Error>>>,
         shutdown_channel: Option<oneshot::Sender<()>>,
-        instrumentation: Arc<std::sync::Mutex<DynInstrumentation>>,
+        instrumentation: Arc<std::sync::Mutex<Option<Box<dyn Instrumentation>>>>,
     ) -> ConnectionResult<Self> {
         let mut conn = Self {
             conn: Arc::new(conn),

src/pooled_connection/mod.rs

@@ -5,10 +5,11 @@
 //! * [deadpool](self::deadpool)
 //! * [bb8](self::bb8)
 //! * [mobc](self::mobc)
+use crate::statement_cache::CacheSize;
 use crate::{AsyncConnection, SimpleAsyncConnection};
 use crate::{TransactionManager, UpdateAndFetchResults};
 use diesel::associations::HasTable;
-use diesel::connection::{CacheSize, Instrumentation};
+use diesel::connection::Instrumentation;
 use diesel::QueryResult;
 use futures_core::future::BoxFuture;
 use futures_util::FutureExt;