Implement stream listener

.tool-versions

@@ -1 +1 @@
-rust 1.85.1
+rust 1.88.0

Cargo.toml

@@ -1,4 +1,5 @@
 [workspace]
+resolver = "3"
 members = [
     "rt",
     "examples/bank",
@@ -17,5 +18,3 @@ spawned-rt = { version = "0.1.0", path = "rt" }
 spawned-concurrency = { version = "0.1.0", path = "concurrency" }
 tracing = { version = "0.1.41", features = ["log"] }
 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
-
-

concurrency/Cargo.toml

@@ -8,5 +8,8 @@ spawned-rt = { workspace = true }
 tracing = { workspace = true }
 futures = "0.3.1"
 
+[dev-dependencies]
+tokio-stream = { version = "0.1.17" }
+
 [lib]
-path = "./src/lib.rs"
+path = "./src/lib.rs"

concurrency/src/tasks/gen_server.rs

@@ -8,7 +8,7 @@ use crate::error::GenServerError;
 
 #[derive(Debug)]
 pub struct GenServerHandle<G: GenServer + 'static> {
-    pub tx: mpsc::Sender<GenServerInMsg<G>>,
+    pub tx: mpsc::UnboundedSender<GenServerInMsg<G>>,
 }
 
 impl<G: GenServer> Clone for GenServerHandle<G> {
@@ -21,7 +21,7 @@ impl<G: GenServer> Clone for GenServerHandle<G> {
 
 impl<G: GenServer> GenServerHandle<G> {
     pub(crate) fn new(initial_state: G::State) -> Self {
-        let (tx, mut rx) = mpsc::channel::<GenServerInMsg<G>>();
+        let (tx, mut rx) = mpsc::unbounded_channel::<GenServerInMsg<G>>();
         let handle = GenServerHandle { tx };
         let mut gen_server: G = GenServer::new();
         let handle_clone = handle.clone();
@@ -39,7 +39,7 @@ impl<G: GenServer> GenServerHandle<G> {
     }
 
     pub(crate) fn new_blocking(initial_state: G::State) -> Self {
-        let (tx, mut rx) = mpsc::channel::<GenServerInMsg<G>>();
+        let (tx, mut rx) = mpsc::unbounded_channel::<GenServerInMsg<G>>();
         let handle = GenServerHandle { tx };
         let mut gen_server: G = GenServer::new();
         let handle_clone = handle.clone();
@@ -58,7 +58,7 @@ impl<G: GenServer> GenServerHandle<G> {
         handle_clone
     }
 
-    pub fn sender(&self) -> mpsc::Sender<GenServerInMsg<G>> {
+    pub fn sender(&self) -> mpsc::UnboundedSender<GenServerInMsg<G>> {
         self.tx.clone()
     }
 
@@ -131,7 +131,7 @@ where
     fn run(
         &mut self,
         handle: &GenServerHandle<Self>,
-        rx: &mut mpsc::Receiver<GenServerInMsg<Self>>,
+        rx: &mut mpsc::UnboundedReceiver<GenServerInMsg<Self>>,
         state: Self::State,
     ) -> impl Future<Output = Result<(), GenServerError>> + Send {
         async {
@@ -162,7 +162,7 @@ where
     fn main_loop(
         &mut self,
         handle: &GenServerHandle<Self>,
-        rx: &mut mpsc::Receiver<GenServerInMsg<Self>>,
+        rx: &mut mpsc::UnboundedReceiver<GenServerInMsg<Self>>,
         mut state: Self::State,
     ) -> impl Future<Output = Result<(), GenServerError>> + Send {
         async {
@@ -181,7 +181,7 @@ where
     fn receive(
         &mut self,
         handle: &GenServerHandle<Self>,
-        rx: &mut mpsc::Receiver<GenServerInMsg<Self>>,
+        rx: &mut mpsc::UnboundedReceiver<GenServerInMsg<Self>>,
         state: Self::State,
     ) -> impl Future<Output = Result<(Self::State, bool), GenServerError>> + Send {
         async move {

concurrency/src/tasks/mod.rs

@@ -3,11 +3,15 @@
 
 mod gen_server;
 mod process;
+mod stream;
 mod time;
 
+#[cfg(test)]
+mod stream_tests;
 #[cfg(test)]
 mod timer_tests;
 
 pub use gen_server::{CallResponse, CastResponse, GenServer, GenServerHandle, GenServerInMsg};
 pub use process::{send, Process, ProcessInfo};
+pub use stream::spawn_listener;
 pub use time::{send_after, send_interval};

concurrency/src/tasks/process.rs

@@ -6,12 +6,12 @@ use std::future::Future;
 
 #[derive(Debug)]
 pub struct ProcessInfo<T> {
-    pub tx: mpsc::Sender<T>,
+    pub tx: mpsc::UnboundedSender<T>,
     pub handle: JoinHandle<()>,
 }
 
 impl<T> ProcessInfo<T> {
-    pub fn sender(&self) -> mpsc::Sender<T> {
+    pub fn sender(&self) -> mpsc::UnboundedSender<T> {
         self.tx.clone()
     }
 
@@ -26,7 +26,7 @@ where
 {
     fn spawn(mut self) -> impl Future<Output = ProcessInfo<T>> + Send {
         async {
-            let (tx, mut rx) = mpsc::channel::<T>();
+            let (tx, mut rx) = mpsc::unbounded_channel::<T>();
             let tx_clone = tx.clone();
             let handle = rt::spawn(async move {
                 self.run(&tx_clone, &mut rx).await;
@@ -37,8 +37,8 @@ where
 
     fn run(
         &mut self,
-        tx: &mpsc::Sender<T>,
-        rx: &mut mpsc::Receiver<T>,
+        tx: &mpsc::UnboundedSender<T>,
+        rx: &mut mpsc::UnboundedReceiver<T>,
     ) -> impl Future<Output = ()> + Send {
         async {
             self.init(tx).await;
@@ -48,8 +48,8 @@ where
 
     fn main_loop(
         &mut self,
-        tx: &mpsc::Sender<T>,
-        rx: &mut mpsc::Receiver<T>,
+        tx: &mpsc::UnboundedSender<T>,
+        rx: &mut mpsc::UnboundedReceiver<T>,
     ) -> impl Future<Output = ()> + Send {
         async {
             loop {
@@ -66,14 +66,14 @@ where
         false
     }
 
-    fn init(&mut self, _tx: &mpsc::Sender<T>) -> impl Future<Output = ()> + Send {
+    fn init(&mut self, _tx: &mpsc::UnboundedSender<T>) -> impl Future<Output = ()> + Send {
         async {}
     }
 
     fn receive(
         &mut self,
-        tx: &mpsc::Sender<T>,
-        rx: &mut mpsc::Receiver<T>,
+        tx: &mpsc::UnboundedSender<T>,
+        rx: &mut mpsc::UnboundedReceiver<T>,
     ) -> impl std::future::Future<Output = T> + Send {
         async {
             match rx.recv().await {
@@ -83,10 +83,14 @@ where
         }
     }
 
-    fn handle(&mut self, message: T, tx: &mpsc::Sender<T>) -> impl Future<Output = T> + Send;
+    fn handle(
+        &mut self,
+        message: T,
+        tx: &mpsc::UnboundedSender<T>,
+    ) -> impl Future<Output = T> + Send;
 }
 
-pub fn send<T>(tx: &mpsc::Sender<T>, message: T)
+pub fn send<T>(tx: &mpsc::UnboundedSender<T>, message: T)
 where
     T: Send,
 {

concurrency/src/tasks/stream.rs

@@ -0,0 +1,50 @@
+use crate::tasks::{GenServer, GenServerHandle};
+use futures::{Stream, StreamExt};
+use spawned_rt::tasks::{CancellationToken, JoinHandle};
+
+/// Spawns a listener that listens to a stream and sends messages to a GenServer.
+///
+/// Items sent through the stream are required to be wrapped in a Result type.
+///
+/// This function returns a handle to the spawned task and a cancellation token
+/// to stop it.
+pub fn spawn_listener<T, F, S, I, E>(
+    mut handle: GenServerHandle<T>,
+    message_builder: F,
+    mut stream: S,
+) -> (JoinHandle<()>, CancellationToken)
+where
+    T: GenServer + 'static,
+    F: Fn(I) -> T::CastMsg + Send + 'static,
+    I: Send,
+    E: std::fmt::Debug + Send,
+    S: Unpin + Send + Stream<Item = Result<I, E>> + 'static,
+{
+    let cancelation_token = CancellationToken::new();
+    let cloned_token = cancelation_token.clone();
+    let join_handle = spawned_rt::tasks::spawn(async move {
+        loop {
+            if cloned_token.is_cancelled() {
+                tracing::trace!("Received signal to stop listener, stopping");
+                break;
+            }
+
+            match stream.next().await {
+                Some(res) => match res {
+                    Ok(i) => {
+                        let _ = handle.cast(message_builder(i)).await;
+                    }
+                    Err(e) => {
+                        tracing::trace!("Received Error in msg {e:?}");
+                        break;
+                    }
+                },
+                None => {
+                    tracing::trace!("Stream finished");
+                    break;
+                }
+            }
+        }
+    });
+    (join_handle, cancelation_token)
+}