Refactor: Improve async runtime compatibility for ODBC

Co-authored-by: contact <[email protected]>

Author: cursoragent

sqlx-core/src/odbc/blocking.rs

@@ -6,6 +6,14 @@ where
     R: Send + 'static,
     F: FnOnce() -> Result<R, Error> + Send + 'static,
 {
-    let res = spawn_blocking(f).await.map_err(|_| Error::WorkerCrashed)?;
-    res
+    #[cfg(feature = "_rt-tokio")]
+    {
+        let join_result = spawn_blocking(f).await.map_err(|_| Error::WorkerCrashed)?;
+        join_result
+    }
+
+    #[cfg(feature = "_rt-async-std")]
+    {
+        spawn_blocking(f).await
+    }
 }

sqlx-core/src/odbc/connection/inner.rs

@@ -1,5 +1,7 @@
 use crate::error::Error;
-use crate::odbc::{OdbcArgumentValue, OdbcArguments, OdbcColumn, OdbcQueryResult, OdbcRow, OdbcTypeInfo};
+use crate::odbc::{
+    OdbcArgumentValue, OdbcArguments, OdbcColumn, OdbcQueryResult, OdbcRow, OdbcTypeInfo,
+};
 use either::Either;
 use flume::{SendError, Sender};
 use odbc_api::handles::StatementImpl;
@@ -105,7 +107,9 @@ where
     C: ResultSetMetadata,
 {
     let count = cursor.num_result_cols().unwrap_or(0);
-    (1..=count).map(|i| create_column(cursor, i as u16)).collect()
+    (1..=count)
+        .map(|i| create_column(cursor, i as u16))
+        .collect()
 }
 
 fn create_column<C>(cursor: &mut C, index: u16) -> OdbcColumn
@@ -177,7 +181,9 @@ fn collect_column_value(
         | DataType::Bit => extract_int(row, col_idx, &type_info)?,
 
         DataType::Real => extract_float::<f32>(row, col_idx, &type_info)?,
-        DataType::Float { .. } | DataType::Double => extract_float::<f64>(row, col_idx, &type_info)?,
+        DataType::Float { .. } | DataType::Double => {
+            extract_float::<f64>(row, col_idx, &type_info)?
+        }
 
         DataType::Char { .. }
         | DataType::Varchar { .. }
@@ -195,10 +201,12 @@ fn collect_column_value(
             extract_binary(row, col_idx, &type_info)?
         }
 
-        DataType::Unknown | DataType::Other { .. } => match extract_text(row, col_idx, &type_info) {
-            Ok(v) => v,
-            Err(_) => extract_binary(row, col_idx, &type_info)?,
-        },
+        DataType::Unknown | DataType::Other { .. } => {
+            match extract_text(row, col_idx, &type_info) {
+                Ok(v) => v,
+                Err(_) => extract_binary(row, col_idx, &type_info)?,
+            }
+        }
     };
 
     Ok((type_info, value))
@@ -289,7 +297,11 @@ fn extract_binary(
     let mut buf = Vec::new();
     let is_some = row.get_binary(col_idx, &mut buf)?;
 
-    let (is_null, blob) = if !is_some { (true, None) } else { (false, Some(buf)) };
+    let (is_null, blob) = if !is_some {
+        (true, None)
+    } else {
+        (false, Some(buf))
+    };
 
     Ok(crate::odbc::OdbcValue {
         type_info: type_info.clone(),
@@ -310,4 +322,3 @@ pub fn do_prepare(
     let params = usize::from(prepared.num_params().unwrap_or(0));
     Ok((0, columns, params))
 }
-

sqlx-core/src/odbc/connection/mod.rs

@@ -1,13 +1,13 @@
 use crate::connection::{Connection, LogSettings};
 use crate::error::Error;
+use crate::odbc::blocking::run_blocking;
 use crate::odbc::{Odbc, OdbcArguments, OdbcColumn, OdbcConnectOptions, OdbcQueryResult, OdbcRow};
 use crate::transaction::Transaction;
 use either::Either;
-use crate::odbc::blocking::run_blocking;
 mod inner;
-use inner::{do_prepare, establish_connection, execute_sql, OdbcConn};
 use futures_core::future::BoxFuture;
 use futures_util::future;
+use inner::{do_prepare, establish_connection, execute_sql, OdbcConn};
 // no direct spawn_blocking here; use run_blocking helper
 use std::sync::{Arc, Mutex};
 
@@ -28,7 +28,8 @@ impl OdbcConnection {
         let conn = run_blocking({
             let options = options.clone();
             move || establish_connection(&options)
-        }).await?;
+        })
+        .await?;
 
         Ok(Self {
             inner: Arc::new(Mutex::new(conn)),
@@ -44,7 +45,8 @@ impl OdbcConnection {
             let conn = inner.lock().unwrap();
             conn.database_management_system_name()
                 .map_err(|e| Error::Protocol(format!("Failed to get DBMS name: {}", e)))
-        }).await
+        })
+        .await
     }
 
     pub(crate) async fn ping_blocking(&mut self) -> Result<(), Error> {
@@ -56,7 +58,8 @@ impl OdbcConnection {
                 Ok(_) => Ok(()),
                 Err(e) => Err(Error::Protocol(format!("Ping failed: {}", e))),
             }
-        }).await
+        })
+        .await
     }
 
     pub(crate) async fn begin_blocking(&mut self) -> Result<(), Error> {
@@ -65,7 +68,8 @@ impl OdbcConnection {
             let conn = inner.lock().unwrap();
             conn.set_autocommit(false)
                 .map_err(|e| Error::Protocol(format!("Failed to begin transaction: {}", e)))
-        }).await
+        })
+        .await
     }
 
     pub(crate) async fn commit_blocking(&mut self) -> Result<(), Error> {
@@ -75,7 +79,8 @@ impl OdbcConnection {
             conn.commit()
                 .and_then(|_| conn.set_autocommit(true))
                 .map_err(|e| Error::Protocol(format!("Failed to commit transaction: {}", e)))
-        }).await
+        })
+        .await
     }
 
     pub(crate) async fn rollback_blocking(&mut self) -> Result<(), Error> {
@@ -85,7 +90,8 @@ impl OdbcConnection {
             conn.rollback()
                 .and_then(|_| conn.set_autocommit(true))
                 .map_err(|e| Error::Protocol(format!("Failed to rollback transaction: {}", e)))
-        }).await
+        })
+        .await
     }
 
     pub(crate) async fn execute_stream(
@@ -102,7 +108,8 @@ impl OdbcConnection {
                 let _ = tx.send(Err(e));
             }
             Ok(())
-        }).await?;
+        })
+        .await?;
         Ok(rx)
     }
 

sqlx-core/src/odbc/mod.rs

@@ -24,6 +24,7 @@
 use crate::executor::Executor;
 
 mod arguments;
+mod blocking;
 mod column;
 mod connection;
 mod database;
@@ -36,7 +37,6 @@ mod transaction;
 mod type_info;
 pub mod types;
 mod value;
-mod blocking;
 
 pub use arguments::{OdbcArgumentValue, OdbcArguments};
 pub use column::OdbcColumn;

sqlx-rt/src/rt_async_std.rs

@@ -1,8 +1,8 @@
 pub use async_std::{
     self, fs, future::timeout, io::prelude::ReadExt as AsyncReadExt,
     io::prelude::WriteExt as AsyncWriteExt, io::Read as AsyncRead, io::Write as AsyncWrite,
-    net::TcpStream, sync::Mutex as AsyncMutex, task::sleep, task::spawn, task::yield_now,
-    task::spawn_blocking,
+    net::TcpStream, sync::Mutex as AsyncMutex, task::sleep, task::spawn, task::spawn_blocking,
+    task::yield_now,
 };
 
 #[cfg(unix)]

sqlx-rt/src/rt_tokio.rs

@@ -1,8 +1,7 @@
 pub use tokio::{
     self, fs, io::AsyncRead, io::AsyncReadExt, io::AsyncWrite, io::AsyncWriteExt, io::ReadBuf,
-    net::TcpStream, runtime::Handle, sync::Mutex as AsyncMutex, task::spawn, task::yield_now,
-    task::spawn_blocking,
-    time::sleep, time::timeout,
+    net::TcpStream, runtime::Handle, sync::Mutex as AsyncMutex, task::spawn, task::spawn_blocking,
+    task::yield_now, time::sleep, time::timeout,
 };
 
 #[cfg(unix)]