Added send_interval and cancellation behavior

ElFantasma · ElFantasma · commit b39e8d561116 · 2025-06-23T17:42:54.000-03:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -8,7 +8,8 @@ members = [
     "examples/ping_pong",
     "examples/ping_pong_threads",
     "examples/updater",
-    "examples/updater_threads", "examples/blocking_genserver",
+    "examples/updater_threads",
+    "examples/blocking_genserver",
 ]
 
 [workspace.dependencies]
diff --git a/concurrency/src/tasks/gen_server.rs b/concurrency/src/tasks/gen_server.rs
@@ -105,8 +105,8 @@ pub trait GenServer
 where
     Self: Send + Sized,
 {
-    type CallMsg: Send + Sized;
-    type CastMsg: Send + Sized;
+    type CallMsg: Clone + Send + Sized + Sync;
+    type CastMsg: Clone + Send + Sized + Sync;
     type OutMsg: Send + Sized;
     type State: Clone + Send;
     type Error: Debug + Send;
diff --git a/concurrency/src/tasks/mod.rs b/concurrency/src/tasks/mod.rs
@@ -9,4 +9,4 @@ mod time;
 pub use error::GenServerError;
 pub use gen_server::{CallResponse, CastResponse, GenServer, GenServerHandle, GenServerInMsg};
 pub use process::{send, Process, ProcessInfo};
-pub use time::send_after;
+pub use time::{send_after, send_interval};
diff --git a/concurrency/src/tasks/time.rs b/concurrency/src/tasks/time.rs
@@ -1,21 +1,72 @@
+use futures::future::select;
 use std::time::Duration;
 
-use spawned_rt::tasks::{self as rt, JoinHandle};
+use spawned_rt::tasks::{self as rt, CancellationToken, JoinHandle};
 
 use super::{GenServer, GenServerHandle};
 
+pub struct TimerHandle {
+    pub join_handle: JoinHandle<()>,
+    pub cancellation_token: CancellationToken,
+}
+
 // Sends a message after a given period to the specified GenServer. The task terminates
 // once the send has completed
 pub fn send_after<T>(
     period: Duration,
     mut handle: GenServerHandle<T>,
     message: T::CastMsg,
-) -> JoinHandle<()>
+) -> TimerHandle
+where
+    T: GenServer + 'static,
+{
+    let cancellation_token = CancellationToken::new();
+    let cloned_token = cancellation_token.clone();
+    let join_handle = rt::spawn(async move {
+        let _ = select(
+            Box::pin(cloned_token.cancelled()),
+            Box::pin(async {
+                rt::sleep(period).await;
+                let _ = handle.cast(message.clone()).await;
+            }),
+        )
+        .await;
+    });
+    TimerHandle {
+        join_handle,
+        cancellation_token,
+    }
+}
+
+// Sends a message to the specified GenServe repeatedly after `Time` milliseconds.
+pub fn send_interval<T>(
+    period: Duration,
+    mut handle: GenServerHandle<T>,
+    message: T::CastMsg,
+) -> TimerHandle
 where
     T: GenServer + 'static,
 {
-    rt::spawn(async move {
-        rt::sleep(period).await;
-        let _ = handle.cast(message).await;
-    })
+    let cancellation_token = CancellationToken::new();
+    let cloned_token = cancellation_token.clone();
+    let join_handle = rt::spawn(async move {
+        loop {
+            let result = select(
+                Box::pin(cloned_token.cancelled()),
+                Box::pin(async {
+                    rt::sleep(period).await;
+                    let _ = handle.cast(message.clone()).await;
+                }),
+            )
+            .await;
+            match result {
+                futures::future::Either::Left(_) => break,
+                futures::future::Either::Right(_) => (),
+            }
+        }
+    });
+    TimerHandle {
+        join_handle,
+        cancellation_token,
+    }
 }
diff --git a/concurrency/src/threads/gen_server.rs b/concurrency/src/threads/gen_server.rs
@@ -83,8 +83,8 @@ pub trait GenServer
 where
     Self: Send + Sized,
 {
-    type CallMsg: Send + Sized;
-    type CastMsg: Send + Sized;
+    type CallMsg: Clone + Send + Sized;
+    type CastMsg: Clone + Send + Sized;
     type OutMsg: Send + Sized;
     type State: Clone + Send;
     type Error: Debug;
diff --git a/concurrency/src/threads/mod.rs b/concurrency/src/threads/mod.rs
@@ -8,4 +8,4 @@ mod time;
 
 pub use gen_server::{CallResponse, CastResponse, GenServer, GenServerHandle, GenServerInMsg};
 pub use process::{send, Process, ProcessInfo};
-pub use time::send_after;
+pub use time::{send_after, send_interval};
diff --git a/concurrency/src/threads/time.rs b/concurrency/src/threads/time.rs
@@ -1,9 +1,14 @@
 use std::time::Duration;
 
-use spawned_rt::threads::{self as rt, JoinHandle};
+use spawned_rt::threads::{self as rt, CancellationToken, JoinHandle};
 
 use super::{GenServer, GenServerHandle};
 
+pub struct TimerHandle {
+    pub join_handle: JoinHandle<()>,
+    pub cancellation_token: CancellationToken,
+}
+
 // Sends a message after a given period to the specified GenServer. The task terminates
 // once the send has completed
 pub fn send_after<T>(
@@ -19,3 +24,28 @@ where
         let _ = handle.cast(message);
     })
 }
+
+// Sends a message to the specified GenServe repeatedly after `Time` milliseconds.
+pub fn send_interval<T>(
+    period: Duration,
+    mut handle: GenServerHandle<T>,
+    message: T::CastMsg,
+) -> TimerHandle
+where
+    T: GenServer + 'static,
+{
+    let cancellation_token = CancellationToken::new();
+    let mut cloned_token = cancellation_token.clone();
+    let join_handle = rt::spawn(move || loop {
+        rt::sleep(period);
+        if cloned_token.is_cancelled() {
+            break;
+        } else {
+            let _ = handle.cast(message.clone());
+        };
+    });
+    TimerHandle {
+        join_handle,
+        cancellation_token,
+    }
+}
diff --git a/concurrency/src/time.rs b/concurrency/src/time.rs
diff --git a/examples/updater/src/main.rs b/examples/updater/src/main.rs
@@ -14,12 +14,15 @@ use spawned_rt::tasks as rt;
 
 fn main() {
     rt::run(async {
+        tracing::info!("Starting Updater");
         UpdaterServer::start(UpdateServerState {
             url: "https://httpbin.org/ip".to_string(),
             periodicity: Duration::from_millis(1000),
+            timer_token: None,
         });
 
         // giving it some time before ending
         thread::sleep(Duration::from_secs(10));
+        tracing::info!("Updater stopped");
     })
 }
diff --git a/examples/updater/src/server.rs b/examples/updater/src/server.rs
@@ -1,8 +1,9 @@
 use std::time::Duration;
 
 use spawned_concurrency::tasks::{
-    send_after, CallResponse, CastResponse, GenServer, GenServerHandle,
+    send_interval, CallResponse, CastResponse, GenServer, GenServerHandle,
 };
+use spawned_rt::tasks::CancellationToken;
 
 use crate::messages::{UpdaterInMessage as InMessage, UpdaterOutMessage as OutMessage};
 
@@ -12,6 +13,7 @@ type UpdateServerHandle = GenServerHandle<UpdaterServer>;
 pub struct UpdateServerState {
     pub url: String,
     pub periodicity: Duration,
+    pub timer_token: Option<CancellationToken>,
 }
 pub struct UpdaterServer {}
 
@@ -26,13 +28,14 @@ impl GenServer for UpdaterServer {
         Self {}
     }
 
-    // Initializing GenServer to start periodic checks
+    // Initializing GenServer to start periodic checks.
     async fn init(
         &mut self,
         handle: &GenServerHandle<Self>,
-        state: Self::State,
+        mut state: Self::State,
     ) -> Result<Self::State, Self::Error> {
-        send_after(state.periodicity, handle.clone(), InMessage::Check);
+        let timer = send_interval(state.periodicity, handle.clone(), InMessage::Check);
+        state.timer_token = Some(timer.cancellation_token);
         Ok(state)
     }
 
@@ -48,18 +51,16 @@ impl GenServer for UpdaterServer {
     async fn handle_cast(
         &mut self,
         message: Self::CastMsg,
-        handle: &UpdateServerHandle,
+        _handle: &UpdateServerHandle,
         state: Self::State,
     ) -> CastResponse<Self> {
         match message {
             Self::CastMsg::Check => {
-                send_after(state.periodicity, handle.clone(), InMessage::Check);
+                //send_after(state.periodicity, handle.clone(), InMessage::Check);
                 let url = state.url.clone();
                 tracing::info!("Fetching: {url}");
                 let resp = req(url).await;
-
                 tracing::info!("Response: {resp:?}");
-
                 CastResponse::NoReply(state)
             }
         }
diff --git a/rt/Cargo.toml b/rt/Cargo.toml
@@ -5,6 +5,7 @@ edition = "2021"
 
 [dependencies]
 tokio = { version = "1", features = ["full"] }
+tokio-util = { version = "0.7.15" }
 crossbeam = { version = "0.7.3" }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
diff --git a/rt/src/tasks/mod.rs b/rt/src/tasks/mod.rs
@@ -16,6 +16,7 @@ use crate::tracing::init_tracing;
 pub use crate::tasks::tokio::mpsc;
 pub use crate::tasks::tokio::oneshot;
 pub use crate::tasks::tokio::sleep;
+pub use crate::tasks::tokio::CancellationToken;
 pub use crate::tasks::tokio::{spawn, spawn_blocking, JoinHandle, Runtime};
 use std::future::Future;
 
diff --git a/rt/src/tasks/tokio/mod.rs b/rt/src/tasks/tokio/mod.rs
@@ -7,3 +7,4 @@ pub use tokio::{
     task::{spawn, spawn_blocking, JoinHandle},
     time::sleep,
 };
+pub use tokio_util::sync::CancellationToken;
diff --git a/rt/src/threads/mod.rs b/rt/src/threads/mod.rs
@@ -3,6 +3,10 @@
 pub mod mpsc;
 pub mod oneshot;
 
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
 pub use std::{
     future::Future,
     thread::{sleep, spawn, JoinHandle},
@@ -29,3 +33,24 @@ where
 {
     spawn(f)
 }
+
+#[derive(Clone, Default)]
+pub struct CancellationToken {
+    is_cancelled: Arc<AtomicBool>,
+}
+
+impl CancellationToken {
+    pub fn new() -> Self {
+        CancellationToken {
+            is_cancelled: Arc::new(false.into()),
+        }
+    }
+
+    pub fn is_cancelled(&mut self) -> bool {
+        self.is_cancelled.fetch_and(false, Ordering::SeqCst)
+    }
+
+    pub fn cancel(&mut self) {
+        self.is_cancelled.fetch_or(true, Ordering::SeqCst);
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -8,7 +8,8 @@ members = [`
`8`	`8`	`"examples/ping_pong",`
`9`	`9`	`"examples/ping_pong_threads",`
`10`	`10`	`"examples/updater",`
`11`		`- "examples/updater_threads", "examples/blocking_genserver",`
	`11`	`+ "examples/updater_threads",`
	`12`	`+ "examples/blocking_genserver",`
`12`	`13`	`]`
`13`	`14`
`14`	`15`	`[workspace.dependencies]`