Repository: slawlor/ractor
Pull request #6 (2023-01-06 00:42:35)

Support for Erlang-style timers.
The following standard Erlang functionalities from the time model are added in this PR

1. `send_after`
2. `send_interval`
3. `exit_after`

There are other functionalities but they're not generally related to `ractor`

Blocked on merging #5 first, then rebasing this


Patch:
diff --git a/ractor/src/actor/actor_cell.rs b/ractor/src/actor/actor_cell.rs
index f6a4ade..7326faf 100644
--- a/ractor/src/actor/actor_cell.rs
+++ b/ractor/src/actor/actor_cell.rs
@@ -39,6 +39,13 @@ pub enum ActorStatus {
     Stopped = 5u8,
 }
 
+/// Denotes states where operations can continue to interact with an agent
+pub const ACTIVE_STATES: [ActorStatus; 3] = [
+    ActorStatus::Starting,
+    ActorStatus::Running,
+    ActorStatus::Upgrading,
+];
+
 /// The collection of ports an actor needs to listen to
 pub struct ActorPortSet {
     pub(crate) signal_rx: BoundedInputPortReceiver<Signal>,
@@ -189,7 +196,7 @@ impl ActorCell {
         other.inner.tree.remove_parent(self.clone());
         self.inner.tree.remove_child(other);
     }
-    /// Stop this actor, but sending Signal::Exit
+    /// Stop this actor, by sending Signal::Exit
     pub fn stop(&self) {
         // ignore failures, since either the actor is already dead
         // or the channel is full of "signals" which is also fine
diff --git a/ractor/src/lib.rs b/ractor/src/lib.rs
index a8972cb..de7bf29 100644
--- a/ractor/src/lib.rs
+++ b/ractor/src/lib.rs
@@ -19,9 +19,10 @@ pub static ACTOR_ID_ALLOCATOR: AtomicU64 = AtomicU64::new(0u64);
 pub mod actor;
 pub mod port;
 pub mod rpc;
+pub mod time;
 
 // re-exports
-pub use actor::actor_cell::{ActorCell, ActorStatus};
+pub use actor::actor_cell::{ActorCell, ActorStatus, ACTIVE_STATES};
 pub use actor::errors::{ActorErr, MessagingErr, SpawnErr};
 pub use actor::messages::{Signal, SupervisionEvent};
 pub use actor::{Actor, ActorHandler};
diff --git a/ractor/src/time/mod.rs b/ractor/src/time/mod.rs
new file mode 100644
index 0000000..130d191
--- /dev/null
+++ b/ractor/src/time/mod.rs
@@ -0,0 +1,67 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! Timers for posting to actors periodically
+
+use tokio::{task::JoinHandle, time::Duration};
+
+use crate::{ActorCell, ActorHandler, Message, MessagingErr, ACTIVE_STATES};
+
+#[cfg(test)]
+mod tests;
+
+/// Sends a message to a given actor repeatedly after a specifid time
+/// using the provided message generation function
+///
+/// Returns: The [JoinHandle] which represents the backgrounded work (can be ignored to
+/// "fire and forget")
+pub fn send_interval<TActor, TMsg, F>(period: Duration, actor: ActorCell, msg: F) -> JoinHandle<()>
+where
+    TActor: ActorHandler<Msg = TMsg>,
+    TMsg: Message,
+    F: Fn() -> TMsg + Send + 'static,
+{
+    tokio::spawn(async move {
+        while ACTIVE_STATES.contains(&actor.get_status()) {
+            tokio::time::sleep(period).await;
+            // if we receive an error trying to send, the channel is closed and we should stop trying
+            // actor died
+            if actor.send_message::<TActor, TMsg>(msg()).is_err() {
+                break;
+            }
+        }
+    })
+}
+
+/// Sends a message after a given period to the specified actor.
+///
+/// Returns: The [JoinHandle] which represents the backgrounded work. Awaiting the handle
+/// will yield the result of the send operation. Can be safely ignored to "fire and forget"
+pub fn send_after<TActor, TMsg, F>(
+    period: Duration,
+    actor: ActorCell,
+    msg: F,
+) -> JoinHandle<Result<(), MessagingErr>>
+where
+    TActor: ActorHandler<Msg = TMsg>,
+    TMsg: Message,
+    F: Fn() -> TMsg + Send + 'static,
+{
+    tokio::spawn(async move {
+        tokio::time::sleep(period).await;
+        actor.send_message::<TActor, TMsg>(msg())
+    })
+}
+
+/// Sends the [crate::Signal::Exit] signal to the actor after a specified duration
+///
+/// Returns: The [JoinHandle] which denotes the backgrounded operation. To cancel the
+/// exit operation, you can abort the handle to cancel the work.
+pub fn exit_after(period: Duration, actor: ActorCell) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        tokio::time::sleep(period).await;
+        actor.stop()
+    })
+}


Test patch:
diff --git a/ractor/src/time/tests.rs b/ractor/src/time/tests.rs
new file mode 100644
index 0000000..ca824b4
--- /dev/null
+++ b/ractor/src/time/tests.rs
@@ -0,0 +1,158 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! Tests of timers
+
+use std::sync::{
+    atomic::{AtomicU8, Ordering},
+    Arc,
+};
+
+use tokio::time::Duration;
+
+use crate::{Actor, ActorCell, ActorHandler};
+
+#[tokio::test]
+async fn test_intervals() {
+    let counter = Arc::new(AtomicU8::new(0u8));
+
+    struct TestActor {
+        counter: Arc<AtomicU8>,
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for TestActor {
+        type Msg = ();
+        type State = Arc<AtomicU8>;
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {
+            self.counter.clone()
+        }
+        async fn handle(
+            &self,
+            _this_actor: ActorCell,
+            _message: Self::Msg,
+            state: &Self::State,
+        ) -> Option<Self::State> {
+            // stop the supervisor, which starts the supervision shutdown of children
+            state.fetch_add(1, Ordering::Relaxed);
+            None
+        }
+    }
+
+    let (actor_ref, actor_handle) = Actor::spawn(
+        None,
+        TestActor {
+            counter: counter.clone(),
+        },
+    )
+    .await
+    .expect("Failed to create test actor");
+
+    let interval_handle = crate::time::send_interval::<TestActor, _, _>(
+        Duration::from_millis(10),
+        actor_ref.clone(),
+        || (),
+    );
+
+    // even though the timer is started, we should be in a "pause" state for 10ms,
+    // therefore the counter should be empty
+    assert_eq!(0, counter.load(Ordering::Relaxed));
+
+    tokio::time::sleep(Duration::from_millis(120)).await;
+    // kill the actor
+    actor_ref.stop();
+
+    tokio::time::sleep(Duration::from_millis(20)).await;
+    // make sure the actor is dead + the interval handle doesn't send again
+    assert!(interval_handle.is_finished());
+    assert!(actor_handle.is_finished());
+
+    // the counter was incremented at least this many times
+    println!("Counter: {}", counter.load(Ordering::Relaxed));
+    assert!(counter.load(Ordering::Relaxed) >= 9);
+}
+
+#[tokio::test]
+async fn test_send_after() {
+    let counter = Arc::new(AtomicU8::new(0u8));
+
+    struct TestActor {
+        counter: Arc<AtomicU8>,
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for TestActor {
+        type Msg = ();
+        type State = Arc<AtomicU8>;
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {
+            self.counter.clone()
+        }
+        async fn handle(
+            &self,
+            _this_actor: ActorCell,
+            _message: Self::Msg,
+            state: &Self::State,
+        ) -> Option<Self::State> {
+            // stop the supervisor, which starts the supervision shutdown of children
+            state.fetch_add(1, Ordering::Relaxed);
+            None
+        }
+    }
+
+    let (actor_ref, actor_handle) = Actor::spawn(
+        None,
+        TestActor {
+            counter: counter.clone(),
+        },
+    )
+    .await
+    .expect("Failed to create test actor");
+
+    let send_after_handle = crate::time::send_after::<TestActor, _, _>(
+        Duration::from_millis(10),
+        actor_ref.clone(),
+        || (),
+    );
+
+    // even though the timer is started, we should be in a "pause" state for 10ms,
+    // therefore the counter should be empty
+    assert_eq!(0, counter.load(Ordering::Relaxed));
+
+    tokio::time::sleep(Duration::from_millis(20)).await;
+    // kill the actor
+    actor_ref.stop();
+
+    tokio::time::sleep(Duration::from_millis(20)).await;
+    // make sure the actor is dead + the interval handle doesn't send again
+    assert!(send_after_handle.is_finished());
+    assert!(actor_handle.is_finished());
+
+    // the counter was incremented at least this many times
+    println!("Counter: {}", counter.load(Ordering::Relaxed));
+    assert_eq!(1, counter.load(Ordering::Relaxed));
+}
+
+#[tokio::test]
+async fn test_exit_after() {
+    struct TestActor;
+
+    #[async_trait::async_trait]
+    impl ActorHandler for TestActor {
+        type Msg = ();
+        type State = ();
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {}
+    }
+
+    let (actor_ref, actor_handle) = Actor::spawn(None, TestActor)
+        .await
+        .expect("Failed to create test actor");
+
+    let exit_handle = crate::time::exit_after(Duration::from_millis(10), actor_ref);
+
+    tokio::time::sleep(Duration::from_millis(20)).await;
+    // make sure the actor is dead + the interval handle doesn't send again
+    assert!(exit_handle.is_finished());
+    assert!(actor_handle.is_finished());
+}