Merge pull request #338 from wprzytula/session-free-waits-for-close

`cass_session_free` waits for session close

Author: wprzytula

scylla-rust-wrapper/src/future.rs

@@ -314,7 +314,7 @@ impl CassFuture {
         CassError::CASS_OK
     }
 
-    fn into_raw(self: Arc<Self>) -> CassOwnedSharedPtr<Self, CMut> {
+    pub(crate) fn into_raw(self: Arc<Self>) -> CassOwnedSharedPtr<Self, CMut> {
         ArcFFI::into_ptr(self)
     }
 

scylla-rust-wrapper/src/session.rs

@@ -105,7 +105,7 @@ impl CassSessionInner {
     }
 
     async fn connect_fut(
-        session_opt: Arc<RwLock<Option<CassSessionInner>>>,
+        session_opt: Arc<RwLock<Option<Self>>>,
         session_builder_fut: impl Future<Output = SessionBuilder>,
         exec_profile_builder_map: HashMap<ExecProfileName, CassExecProfile>,
         host_filter: Arc<dyn HostFilter>,
@@ -163,6 +163,26 @@ impl CassSessionInner {
         });
         Ok(CassResultValue::Empty)
     }
+
+    fn close_fut(session_opt: Arc<RwLock<Option<Self>>>) -> Arc<CassFuture> {
+        CassFuture::new_from_future(
+            async move {
+                let mut session_guard = session_opt.write().await;
+                if session_guard.is_none() {
+                    return Err((
+                        CassError::CASS_ERROR_LIB_UNABLE_TO_CLOSE,
+                        "Already closing or closed".msg(),
+                    ));
+                }
+
+                *session_guard = None;
+
+                Ok(CassResultValue::Empty)
+            },
+            #[cfg(cpp_integration_testing)]
+            None,
+        )
+    }
 }
 
 pub type CassSession = RwLock<Option<CassSessionInner>>;
@@ -352,13 +372,12 @@ pub unsafe extern "C" fn cass_session_execute(
 
     let future = async move {
         let session_guard = session_opt.read().await;
-        if session_guard.is_none() {
+        let Some(cass_session_inner) = session_guard.as_ref() else {
             return Err((
                 CassError::CASS_ERROR_LIB_NO_HOSTS_AVAILABLE,
                 "Session is not connected".msg(),
             ));
-        }
-        let cass_session_inner = session_guard.as_ref().unwrap();
+        };
         let session = &cass_session_inner.session;
 
         let handle = cass_session_inner
@@ -576,7 +595,16 @@ pub unsafe extern "C" fn cass_session_prepare_n(
 
 #[unsafe(no_mangle)]
 pub unsafe extern "C" fn cass_session_free(session_raw: CassOwnedSharedPtr<CassSession, CMut>) {
-    ArcFFI::free(session_raw);
+    let Some(session_opt) = ArcFFI::from_ptr(session_raw) else {
+        // `free()` does nothing on null pointers, so by analogy let's do nothing here.
+        return;
+    };
+
+    let close_fut = CassSessionInner::close_fut(session_opt);
+    close_fut.with_waited_result(|_| ());
+
+    // We don't have to drop the session's Arc explicitly, because it has been moved
+    // into the CassFuture, which is dropped here with the end of the scope.
 }
 
 #[unsafe(no_mangle)]
@@ -588,23 +616,7 @@ pub unsafe extern "C" fn cass_session_close(
         return ArcFFI::null();
     };
 
-    CassFuture::make_raw(
-        async move {
-            let mut session_guard = session_opt.write().await;
-            if session_guard.is_none() {
-                return Err((
-                    CassError::CASS_ERROR_LIB_UNABLE_TO_CLOSE,
-                    "Already closing or closed".msg(),
-                ));
-            }
-
-            *session_guard = None;
-
-            Ok(CassResultValue::Empty)
-        },
-        #[cfg(cpp_integration_testing)]
-        None,
-    )
+    CassSessionInner::close_fut(session_opt).into_raw()
 }
 
 #[unsafe(no_mangle)]