Minor improvements

Author: weiznich

src/lib.rs

@@ -130,7 +130,7 @@ mod transaction_manager;
 pub use self::mysql::AsyncMysqlConnection;
 #[cfg(feature = "mysql")]
 #[doc(inline)]
-pub use self::mysql::CancelToken;
+pub use self::mysql::MysqlCancelToken;
 #[cfg(feature = "postgres")]
 #[doc(inline)]
 pub use self::pg::AsyncPgConnection;

src/mysql/cancel_token.rs

@@ -6,12 +6,12 @@ use crate::mysql::error_helper::ErrorHelper;
 /// The capability to request cancellation of in-progress queries on a
 /// connection.
 #[derive(Clone)]
-pub struct CancelToken {
+pub struct MysqlCancelToken {
     pub(crate) opts: Opts,
     pub(crate) kill_id: u32,
 }
 
-impl CancelToken {
+impl MysqlCancelToken {
     /// Attempts to cancel the in-progress query on the connection associated
     /// with this `CancelToken`.
     ///
@@ -22,7 +22,7 @@ impl CancelToken {
     /// cancellation request will reach the server before the query terminates
     /// normally, or that the connection associated with this token is still
     /// active.
-    pub async fn cancel_query(&self) -> Result<(), diesel::result::ConnectionError> {
+    pub async fn cancel_query(&self) -> diesel::result::ConnectionResult<()> {
         let builder = OptsBuilder::from_opts(self.opts.clone());
 
         let conn = mysql_async::Conn::new(builder).await.map_err(ErrorHelper)?;

src/mysql/mod.rs

@@ -24,7 +24,7 @@ mod error_helper;
 mod row;
 mod serialize;
 
-pub use self::cancel_token::CancelToken;
+pub use self::cancel_token::MysqlCancelToken;
 use self::error_helper::ErrorHelper;
 use self::row::MysqlRow;
 use self::serialize::ToSqlHelper;
@@ -257,11 +257,11 @@ impl AsyncMysqlConnection {
     }
 
     /// Constructs a cancellation token that can later be used to request cancellation of a query running on the connection associated with this client.
-    pub fn cancel_token(&self) -> CancelToken {
+    pub fn cancel_token(&self) -> MysqlCancelToken {
         let kill_id = self.conn.id();
         let opts = self.conn.opts().clone();
 
-        CancelToken { kill_id, opts }
+        MysqlCancelToken { kill_id, opts }
     }
 
     fn with_prepared_statement<'conn, T, F, R>(

tests/lib.rs

@@ -170,14 +170,13 @@ async fn postgres_cancel_token() {
 #[cfg(feature = "mysql")]
 #[tokio::test]
 async fn mysql_cancel_token() {
-    use std::time::Duration;
-
     use diesel::result::{DatabaseErrorKind, Error};
+    use std::time::Duration;
 
     let (sender, receiver) = tokio::sync::oneshot::channel();
 
-    // execute a long-running query on a separate thread
-    let task = tokio::spawn(async move {
+    // execute a long-running query in a separate future
+    let query_future = async move {
         let conn = &mut connection().await;
         let token = conn.cancel_token();
 
@@ -187,28 +186,30 @@ async fn mysql_cancel_token() {
             .unwrap_or_else(|_| panic!("couldn't send token"));
 
         diesel::dsl::sql::<diesel::sql_types::Integer>("SELECT SLEEP(5)")
-            .load::<i32>(conn)
+            .get_result::<i32>(conn)
             .await
-    });
-
-    // wait for the cancellation token to be sent
-    if let Ok(token) = receiver.await {
-        // give the query time to start before invoking the token
-        tokio::time::sleep(Duration::from_millis(500)).await;
-        token.cancel_query().await.unwrap();
-    }
+    };
+    let cancel_future = async move {
+        // wait for the cancellation token to be sent
+        if let Ok(token) = receiver.await {
+            // give the query time to start before invoking the token
+            tokio::time::sleep(Duration::from_millis(500)).await;
+            token.cancel_query().await.unwrap();
+        } else {
+            panic!("Failed to receive cancel token");
+        }
+    };
+
+    let (task, _) = tokio::join!(query_future, cancel_future);
 
     // make sure the query task resulted in a cancellation error or a return value of 1:
-    match task.await.unwrap() {
-        Err(e) => match e {
-            Error::DatabaseError(DatabaseErrorKind::Unknown, v)
-                if v.message() == "Query execution was interrupted" => {}
-            _ => panic!("unexpected error: {:?}", e),
-        },
-        Ok(r) => match r[0] {
-            1 => {}
-            _ => panic!("query completed successfully without cancellation"),
-        },
+    match task {
+        Err(Error::DatabaseError(DatabaseErrorKind::Unknown, v))
+            if v.message() == "Query execution was interrupted" => {}
+        Err(e) => panic!("unexpected error: {:?}", e),
+        // mysql 8.4 returns 1 from a canceled sleep instead of an error
+        Ok(1) => {}
+        Ok(_) => panic!("query completed successfully without cancellation"),
     }
 }