chore: Return a CancellationToken when shutting down an Actor

Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 
 [dependencies]
 tokio = { version = "1.21.2", features = ["full"] }
+tokio-util = "0.7.8"
 theater_derive = { path = "./theater_derive" }
 theater_types = { path = "./theater_types" }
 async-trait = "0.1.58"

src/lib.rs

@@ -1,6 +1,6 @@
 /// Tiny, embeddable actor based system utilities
 mod theater;
-pub use theater::*;
+pub use self::theater::*;
 pub use theater_types::*;
 
 // TODO: revisit derive API later

src/theater.rs

@@ -3,6 +3,7 @@ use std::marker::PhantomData;
 use async_trait::async_trait;
 use theater_types::{Actor, ActorId, ActorLabel, ActorState, Handler, Result};
 use tokio::sync::broadcast::Receiver;
+use tokio_util::sync::CancellationToken;
 
 #[derive(Debug, Clone)]
 pub struct ActorImpl<H, M>
@@ -29,7 +30,7 @@ where
         }
     }
 
-    fn set_early_stop(&mut self, val: bool) {
+    fn _set_early_stop(&mut self, val: bool) {
         self.stop_early = val
     }
 }
@@ -52,7 +53,7 @@ where
         self.handler.status()
     }
 
-    async fn start(&mut self, message_rx: &mut Receiver<M>) -> Result<()> {
+    async fn start(&mut self, message_rx: &mut Receiver<M>) -> Result<CancellationToken> {
         self.handler.set_status(ActorState::Starting);
 
         self.handler.on_start();
@@ -74,14 +75,14 @@ where
                         self.handler.set_status(ActorState::Terminating);
                         self.handler.on_stop();
 
-                        return Ok(());
+                        return Ok(CancellationToken::new());
                     }
                 },
             }
         }
 
         self.handler.set_status(ActorState::Stopped);
 
-        Ok(())
+        Ok(CancellationToken::new())
     }
 }

theater_types/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2021"
 syn = "1.0"
 quote = "1.0"
 tokio = { version = "1.21.2", features = ["full"] }
+tokio-util = "0.7.8"
 anyhow = "1.0.65"
 thiserror = "1.0.37"
 async-trait = "0.1.58"

theater_types/src/lib.rs

@@ -20,7 +20,6 @@ where
     fn set_status(&mut self, state: ActorState);
 
     /// Called every time a message is received by an actor
-    // async fn handle(&mut self, msg: impl std::fmt::Debug + Clone) -> Result<ActorState>;
     async fn handle(&mut self, msg: M) -> Result<ActorState>;
 
     /// Called before starting the message processing loop
@@ -68,5 +67,6 @@ where
     /// Optional human-readable label
     fn label(&self) -> ActorLabel;
     fn status(&self) -> ActorState;
-    async fn start(&mut self, message_rx: &mut Receiver<M>) -> Result<()>;
+    async fn start(&mut self, message_rx: &mut Receiver<M>) -> Result<tokio_util::sync::CancellationToken>;
 }
+