feat: Allow for async row streams to be Send

Author: pacman82

Cargo.lock

@@ -120,6 +120,8 @@ tempfile = "3.21.0"
 criterion = { version = "0.7.0", features = ["html_reports"] }
 tokio = { version = "1.47.1", features = ["rt", "macros", "time"] }
 stdext = "0.3.3"                                                    # Used for function_name macro to generate unique table names for tests
+tokio-stream = "0.1.17"
+async-stream = "0.3.6"
 
 
 [[bench]]

odbc-api/Cargo.toml

@@ -2,7 +2,7 @@ use std::mem::transmute;
 
 use crate::{
     CursorImpl, CursorPolling, Error, ParameterCollectionRef, Sleep,
-    handles::{AsStatementRef, SqlText, Statement},
+    handles::{AsStatementRef, SqlText, Statement, StatementRef},
     parameter::Blob,
     sleep::wait_for,
 };
@@ -165,10 +165,7 @@ where
         if need_data {
             // Check if any delayed parameters have been bound which stream data to the database at
             // statement execution time. Loops over each bound stream.
-            while let Some(blob_ptr) = stmt.param_data().into_result(&stmt)? {
-                // The safe interfaces currently exclusively bind pointers to `Blob` trait objects
-                let blob_ptr: *mut &mut dyn Blob = transmute(blob_ptr);
-                let blob_ref = &mut *blob_ptr;
+            while let Some(blob_ref) = next_blob_param(&mut stmt)? {
                 // Loop over all batches within each blob
                 while let Some(batch) = blob_ref.next_batch().map_err(Error::FailedReadingInput)? {
                     let result = wait_for(|| stmt.put_binary_batch(batch), &mut sleep).await;
@@ -191,6 +188,20 @@ where
     }
 }
 
+unsafe fn next_blob_param<'a>(
+    stmt: &mut StatementRef<'a>,
+) -> Result<Option<&'a mut dyn Blob>, Error> {
+    let maybe_ptr = stmt.param_data().into_result(stmt)?;
+    if let Some(blob_ptr) = maybe_ptr {
+        // The safe interfaces currently exclusively bind pointers to `Blob` trait objects
+        let blob_ptr: *mut &mut dyn Blob = unsafe { std::mem::transmute(blob_ptr) };
+        let blob_ref = unsafe { &mut *blob_ptr };
+        Ok(Some(*blob_ref))
+    } else {
+        Ok(None)
+    }
+}
+
 /// Shared implementation for executing a columns query between [`crate::Connection`] and
 /// [`crate::Preallocated`].
 pub fn execute_columns<S>(

odbc-api/src/execute.rs

@@ -244,7 +244,7 @@
 //! fn insert_image_to_db(
 //!     conn: &Connection<'_>,
 //!     id: &str,
-//!     image_data: impl BufRead) -> Result<(), Error>
+//!     image_data: impl BufRead + Send) -> Result<(), Error>
 //! {
 //!     const MAX_IMAGE_SIZE: usize = 4 * 1024 * 1024;
 //!     let mut blob = BlobRead::with_upper_bound(image_data, MAX_IMAGE_SIZE);

odbc-api/src/parameter.rs

@@ -12,14 +12,14 @@ use std::{
     path::Path,
 };
 
-/// A `Blob` can stream its contents to the database batch by batch and may therefore be used to
+/// A [`Blob`] can stream its contents to the database batch by batch and may therefore be used to
 /// transfer large amounts of data, exceeding the drivers capabilities for normal input parameters.
 ///
 /// # Safety
 ///
 /// If a hint is implemented for `blob_size` it must be accurate before the first call to
 /// `next_batch`.
-pub unsafe trait Blob: HasDataType {
+pub unsafe trait Blob: HasDataType + Send {
     /// CData type of the binary data returned in the batches. Likely to be either
     /// [`crate::sys::CDataType::Binary`], [`crate::sys::CDataType::Char`] or
     /// [`crate::sys::CDataType::WChar`].
@@ -266,7 +266,7 @@ impl<R> BlobRead<R> {
     /// fn insert_image_to_db(
     ///     conn: &Connection<'_>,
     ///     id: &str,
-    ///     image_data: impl BufRead) -> Result<(), Error>
+    ///     image_data: impl BufRead + Send) -> Result<(), Error>
     /// {
     ///     const MAX_IMAGE_SIZE: usize = 4 * 1024 * 1024;
     ///     let mut blob = BlobRead::with_upper_bound(image_data, MAX_IMAGE_SIZE);
@@ -360,7 +360,7 @@ where
 
 unsafe impl<R> Blob for BlobRead<R>
 where
-    R: BufRead,
+    R: BufRead + Send,
 {
     fn c_data_type(&self) -> CDataType {
         CDataType::Binary

odbc-api/src/parameter/blob.rs

@@ -1,6 +1,7 @@
 mod common;
 mod connection_strings;
 
+use async_stream::stream;
 use odbc_sys::{Date, Time};
 use stdext::function_name;
 use sys::{CDataType, NULL_DATA, Numeric, Pointer, SqlDataType, Timestamp};
@@ -34,6 +35,7 @@ use odbc_api::{
     },
     sys,
 };
+use tokio_stream::{Stream, StreamExt as _};
 use widestring::Utf16String;
 
 use std::{
@@ -6063,6 +6065,39 @@ fn fetch_decimal_as_numeric_struct_using_bind_col(profile: &Profile) {
     assert_eq!(0, target.val[2]);
 }
 
+#[test_case(MSSQL; "Microsoft SQL Server")]
+#[test_case(MARIADB; "Maria DB")]
+#[test_case(SQLITE_3; "SQLite 3")]
+#[test_case(POSTGRES; "PostgreSQL")]
+#[tokio::test]
+async fn async_stream_of_rows_from_other_thread(profile: &Profile) {
+    let table_name = table_name!();
+    let (conn, table) = Given::new(&table_name)
+        .column_types(&["INT"])
+        .values_by_column(&[&[Some("42")]])
+        .build(profile)
+        .unwrap();
+
+    // When
+    fn stream_of_send_rows(
+        connection: Connection<'static>,
+        query: String,
+    ) -> impl Stream<Item = (i32,)> + Send {
+        let stmt = connection.into_preallocated().unwrap();
+        let mut stmt = stmt.into_polling().unwrap();
+        stream! {
+            let sleep = || tokio::time::sleep(Duration::from_millis(10));
+            let _ = stmt.execute(&query, (), sleep).await;
+            yield (42, )
+        }
+    }
+
+    // Then
+    let stream = stream_of_send_rows(conn, table.sql_all_ordered_by_id());
+    let rows = stream.collect::<Vec<_>>().await;
+    assert_eq!([(42i32,)].as_slice(), rows)
+}
+
 /// Learning test to see how scrolling cursors behave
 #[test_case(MSSQL; "Microsoft SQL Server")]
 #[test_case(MARIADB; "Maria DB")]