feat: shared_connection_into_cursor allows cursors to share ownership

of the same connection across threads.

Author: pacman82

odbc-api/src/connection.rs

@@ -311,8 +311,10 @@ impl<'c> Connection<'c> {
         let Some(cursor) = maybe_cursor else {
             return Ok(None);
         };
+        // Deconstrurt the cursor and construct which only borrows the connection and construct a new
+        // one which takes ownership of the instead.
         let stmt_ptr = cursor.into_stmt().into_sys();
-        // Safety: The connection is the parent of the statement referenced by `stmt_ptr`.
+        // Safe: The connection is the parent of the statement referenced by `stmt_ptr`.
         let stmt = unsafe { StatementConnection::new(stmt_ptr, Arc::clone(&self)) };
         // Safe: `stmt` is valid and in cursor state.
         let cursor = unsafe { CursorImpl::new(stmt) };

odbc-api/src/lib.rs

@@ -52,7 +52,7 @@ pub use self::{
     prepared::Prepared,
     result_set_metadata::ResultSetMetadata,
     sleep::Sleep,
-    sync_connection::SharedConnection,
+    sync_connection::{SharedConnection, shared_connection_into_cursor},
 };
 
 /// Reexports `odbc-sys` as sys to enable applications to always use the same version as this

odbc-api/src/sync_connection.rs

@@ -1,6 +1,9 @@
 use std::sync::{Arc, Mutex};
 
-use crate::{Connection, handles::StatementParent};
+use crate::{
+    Connection, CursorImpl, Error, ParameterCollectionRef,
+    handles::{StatementConnection, StatementParent},
+};
 
 /// A convinient type alias in case you want to use a connection from multiple threads which share
 /// ownership of it.
@@ -11,3 +14,53 @@ pub type SharedConnection<'env> = Arc<Mutex<Connection<'env>>>;
 /// Connection is guaranteed to be alive and in connected state for the lifetime of
 /// [`SharedConnection`].
 unsafe impl StatementParent for SharedConnection<'_> {}
+
+/// Similar to [`crate::Connection::into_cursor`], yet it operates on an `Arc<Mutex<Connection>>`.
+/// `Arc<Connection>` can be used if you want shared ownership of connections. However,
+/// `Arc<Connection>` is not `Send` due to `Connection` not being `Sync`. So sometimes you may want
+/// to wrap your `Connection` into an `Arc<Mutex<Connection>>` to allow shared ownership of the
+/// connection across threads. This function allows you to create a cursor from such a shared
+/// which also holds a strong reference to it.
+///
+/// # Parameters
+///
+/// * `query`: The text representation of the SQL statement. E.g. "SELECT * FROM my_table;".
+/// * `params`: `?` may be used as a placeholder in the statement text. You can use `()` to
+///   represent no parameters. See the [`crate::parameter`] module level documentation for more
+///   information on how to pass parameters.
+/// * `query_timeout_sec`: Use this to limit the time the query is allowed to take, before
+///   responding with data to the application. The driver may replace the number of seconds you
+///   provide with a minimum or maximum value.
+///
+///   For the timeout to work the driver must support this feature. E.g. PostgreSQL, and Microsoft
+///   SQL Server do, but SQLite or MariaDB do not.
+///
+///   You can specify ``0``, to deactivate the timeout, this is the default. So if you want no
+///   timeout, just leave it at `None`. Only reason to specify ``0`` is if for some reason your
+///   datasource does not have ``0`` as default.
+///
+///   This corresponds to `SQL_ATTR_QUERY_TIMEOUT` in the ODBC C API.
+///
+///   See: <https://learn.microsoft.com/en-us/sql/odbc/reference/syntax/sqlsetstmtattr-function>
+pub fn shared_connection_into_cursor<'env>(
+    connection: SharedConnection<'env>,
+    query: &str,
+    params: impl ParameterCollectionRef,
+    query_timeout_sec: Option<usize>,
+) -> Result<Option<CursorImpl<StatementConnection<SharedConnection<'env>>>>, Error> {
+    let guard = connection
+        .lock()
+        .expect("Shared connection lock must not be poisned");
+    let Some(cursor) = guard.execute(query, params, query_timeout_sec)? else {
+        return Ok(None);
+    };
+    // Deconstrurt the cursor and construct which only borrows the connection and construct a new
+    // one which takes ownership of the instead.
+    let stmt_ptr = cursor.into_stmt().into_sys();
+    drop(guard);
+    // Safe: The connection is the parent of the statement referenced by `stmt_ptr`.
+    let stmt = unsafe { StatementConnection::new(stmt_ptr, connection) };
+    // Safe: `stmt` is valid and in cursor state.
+    let cursor = unsafe { CursorImpl::new(stmt) };
+    Ok(Some(cursor))
+}

odbc-api/tests/integration.rs

@@ -31,7 +31,7 @@ use odbc_api::{
         Blob, BlobRead, BlobSlice, InputParameter, VarBinaryArray, VarCharArray, VarCharSlice,
         VarCharSliceMut, VarWCharArray, WithDataType,
     },
-    sys,
+    shared_connection_into_cursor, sys,
 };
 use widestring::Utf16String;
 
@@ -42,7 +42,7 @@ use std::{
     num::NonZeroUsize,
     ptr::null_mut,
     str,
-    sync::Arc,
+    sync::{Arc, Mutex},
     thread,
     time::{Duration, Instant},
 };
@@ -604,6 +604,42 @@ fn shared_ownership_of_connections_by_statement(profile: &Profile) {
     assert_eq!(expected, cursor_to_string(cursor));
 }
 
+#[test_case(MSSQL; "Microsoft SQL Server")]
+#[test_case(MARIADB; "Maria DB")]
+#[test_case(SQLITE_3; "SQLite 3")]
+#[test_case(POSTGRES; "PostgreSQL")]
+fn share_connections_with_statement_in_other_thread(profile: &Profile) {
+    // Given
+    let table_name = table_name!();
+    let (conn, table) = Given::new(&table_name)
+        .column_types(&["INT"])
+        .values_by_column(&[&[Some("42")]])
+        .build(profile)
+        .unwrap();
+
+    // When
+    let conn = Arc::new(Mutex::new(conn));
+    let mut cursor =
+        shared_connection_into_cursor(conn.clone(), &table.sql_all_ordered_by_id(), (), None)
+            .unwrap()
+            .unwrap();
+    let other_thread = thread::spawn(move || {
+        let mut i = 0i32;
+        cursor
+            .next_row()
+            .unwrap()
+            .unwrap()
+            .get_data(1, &mut i)
+            .unwrap();
+        i
+    });
+    drop(conn);
+    let answer = other_thread.join().unwrap();
+
+    // Then
+    assert_eq!(42, answer);
+}
+
 /// Strong exception safety for `into_cursor`. Our first query will fail, because it will query a
 /// non-existing table, but our second one using the same connection will succeed. This is one
 /// scenario in which it is useful not to "swallow" the connection in case of an error.