Add a wait() operator for an actor (#304)

This allows someone to wait for the actor's exit without waiting on the JoinHandle

ractor/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ractor"
-version = "0.14.1"
+version = "0.14.2"
 authors = ["Sean Lawlor", "Evan Au", "Dillon George"]
 description = "A actor framework for Rust"
 documentation = "https://docs.rs/ractor"

ractor/src/actor/actor_cell.rs

@@ -451,6 +451,25 @@ impl ActorCell {
         }
     }
 
+    /// Wait for the actor to exit, optionally within a timeout
+    ///
+    /// * `timeout`: If supplied, the amount of time to wait before
+    ///   returning an error and cancelling the wait future.
+    ///
+    /// IMPORTANT: If the timeout is hit, the actor is still running.
+    /// You should wait again for its exit.
+    pub async fn wait(
+        &self,
+        timeout: Option<crate::concurrency::Duration>,
+    ) -> Result<(), crate::concurrency::Timeout> {
+        if let Some(to) = timeout {
+            crate::concurrency::timeout(to, self.inner.wait()).await
+        } else {
+            self.inner.wait().await;
+            Ok(())
+        }
+    }
+
     /// Send a supervisor event to the supervisory port
     ///
     /// * `message` - The [SupervisionEvent] to send to the supervisory port

ractor/src/actor/actor_properties.rs

@@ -225,6 +225,12 @@ impl ActorProperties {
         Ok(())
     }
 
+    /// Wait for the actor to exit
+    pub(crate) async fn wait(&self) {
+        let rx = self.wait_handler.notified();
+        rx.await;
+    }
+
     /// Send the kill signal, threading in a OneShot sender which notifies when the shutdown is completed
     pub(crate) async fn send_signal_and_wait(
         &self,
@@ -239,5 +245,9 @@ impl ActorProperties {
 
     pub(crate) fn notify_stop_listener(&self) {
         self.wait_handler.notify_waiters();
+        // make sure that any future caller immediately returns by pre-storing
+        // a notify permit (i.e. the actor stops, but you are only start waiting
+        // after the actor has already notified it's dead.)
+        self.wait_handler.notify_one();
     }
 }

ractor/src/actor/tests/mod.rs

@@ -1156,3 +1156,44 @@ async fn runtime_message_typing() {
     actor.stop(None);
     handle.await.unwrap();
 }
+
+#[crate::concurrency::test]
+#[tracing_test::traced_test]
+async fn wait_for_death() {
+    struct TestActor;
+
+    #[cfg_attr(feature = "async-trait", crate::async_trait)]
+    impl Actor for TestActor {
+        type Msg = EmptyMessage;
+        type Arguments = ();
+        type State = ();
+
+        async fn pre_start(
+            &self,
+            _this_actor: crate::ActorRef<Self::Msg>,
+            _: (),
+        ) -> Result<Self::State, ActorProcessingErr> {
+            Ok(())
+        }
+
+        async fn post_stop(
+            &self,
+            _myself: ActorRef<Self::Msg>,
+            _: &mut Self::State,
+        ) -> Result<(), ActorProcessingErr> {
+            crate::concurrency::sleep(Duration::from_millis(10)).await;
+            Ok(())
+        }
+    }
+
+    let (actor, handle) = Actor::spawn(None, TestActor, ())
+        .await
+        .expect("Failed to start test actor");
+
+    actor.stop(None);
+    assert!(actor.wait(Some(Duration::from_millis(100))).await.is_ok());
+
+    // cleanup
+    actor.stop(None);
+    handle.await.unwrap();
+}

ractor_cluster/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ractor_cluster"
-version = "0.14.1"
+version = "0.14.2"
 authors = ["Sean Lawlor <slawlor>"]
 description = "Distributed cluster environment of Ractor actors"
 documentation = "https://docs.rs/ractor"

ractor_cluster_derive/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ractor_cluster_derive"
-version = "0.14.1"
+version = "0.14.2"
 authors = ["Sean Lawlor <slawlor>"]
 description = "Derives for ractor_cluster"
 license = "MIT"