refactor(odbc): simplify execute_stream implementation

This commit refactors the execute_stream method in OdbcConnection to directly return the receiver from the execute_stream function, eliminating unnecessary complexity. Additionally, the method signature is updated to reflect the change in return type, enhancing clarity in the codebase.

Author: lovasoa

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

@@ -20,15 +20,8 @@ impl<'c> Executor<'c> for &'c mut OdbcConnection {
         'c: 'e,
         E: Execute<'q, Self::Database> + 'q,
     {
-        let sql = query.sql().to_string();
         let args = query.take_arguments();
-        Box::pin(try_stream! {
-            let rx = self.execute_stream(&sql, args).await?;
-            while let Ok(item) = rx.recv_async().await {
-                r#yield!(item?);
-            }
-            Ok(())
-        })
+        Box::pin(self.execute_stream(query.sql(), args).into_stream())
     }
 
     fn fetch_optional<'e, 'q: 'e, E>(

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

@@ -138,29 +138,29 @@ impl OdbcConnection {
         .await
     }
 
-    pub(crate) async fn execute_stream(
+    /// Launches a background task to execute the SQL statement and send the results to the returned channel.
+    pub(crate) fn execute_stream(
         &mut self,
         sql: &str,
         args: Option<OdbcArguments>,
-    ) -> Result<flume::Receiver<Result<Either<OdbcQueryResult, OdbcRow>, Error>>, Error> {
+    ) -> flume::Receiver<Result<Either<OdbcQueryResult, OdbcRow>, Error>> {
         let (tx, rx) = flume::bounded(64);
 
-        // !!TODO!!!:  Put back the prepared statement after usage
         let maybe_prepared = if let Some(prepared) = self.stmt_cache.get(sql) {
             MaybePrepared::Prepared(Arc::clone(prepared))
         } else {
             MaybePrepared::NotPrepared(sql.to_string())
         };
 
-        self.with_conn("execute_stream", move |conn| {
-            if let Err(e) = execute_sql(conn, maybe_prepared, args, &tx) {
+        let conn = Arc::clone(&self.conn);
+        sqlx_rt::spawn(sqlx_rt::spawn_blocking(move || {
+            let mut conn = conn.lock().expect("failed to lock connection");
+            if let Err(e) = execute_sql(&mut conn, maybe_prepared, args, &tx) {
                 let _ = tx.send(Err(e));
             }
-            Ok(())
-        })
-        .await?;
+        }));
 
-        Ok(rx)
+        rx
     }
 
     pub(crate) async fn clear_cached_statements(&mut self) -> Result<(), Error> {