Merge pull request #3 from vmenge/stream

adds stream spawning functionality

core/Cargo.toml

@@ -12,6 +12,7 @@ readme = "../README.md"
 async-trait = { version = "0.1.73" }
 flume = "0.11.0"
 tokio = { version = "1.35.1", features = ["macros", "rt", "sync", "time"] }
+futures = "0.3.31"
 
 [dev-dependencies]
 tokio = { version = "1.35.1", features = [

core/src/lib.rs

@@ -1,17 +1,20 @@
 use async_trait::async_trait;
 use flume::{Receiver, Sender};
-use std::{any::Any, collections::HashMap, time::Duration};
+use futures::Stream;
+use std::{any::Any, collections::HashMap, future::Future, time::Duration};
 use tokio::{
     task,
     time::{self},
 };
 
 mod exit;
 mod req_res;
+mod stream;
 mod supervision;
 
 pub use exit::*;
 pub use req_res::*;
+pub use stream::*;
 pub use supervision::*;
 
 /// A thin abstraction over tokio tasks and flume channels, allowing for easy message passing
@@ -242,6 +245,16 @@ impl<Msg> Handle<Msg> {
         res.recv().await
     }
 
+    pub async fn reqw<F, Req, Res>(&self, to_req: F, req: Req) -> Result<Res, ReqErr>
+    where
+        F: Fn(Request<Req, Res>) -> Msg
+    {
+        let (req, res) = req_res(req);
+        let msg = to_req(req);
+        self.send(msg);
+        res.recv().await
+    }
+
     /// Sends a request to the `Process` as long as its messages implements `From<Request<Req,Res>>`.
     ///
     /// Fails if response is not sent back within the given `Duration`.
@@ -253,6 +266,16 @@ impl<Msg> Handle<Msg> {
         self.send(req);
         res.recv_timeout(timeout).await
     }
+
+    pub async fn reqw_timeout<F, Req, Res>(&self, to_req: F, req: Req, timeout: Duration) -> Result<Res, ReqErr>
+    where
+        F: Fn(Request<Req, Res>) -> Msg
+    {
+        let (req, res) = req_res(req);
+        let msg = to_req(req);
+        self.send(msg);
+        res.recv_timeout(timeout).await
+    }
 }
 
 /// The context surrounding the current `Process`.
@@ -338,6 +361,18 @@ where
         handle
     }
 
+    pub fn stream<F, Fut, S, T, E>(&mut self, f: F) -> StreamBuilder<'_, P, F, Fut, S, T, E, NoSink>
+    where
+        F: Fn() -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = S> + Send + Sync + 'static,
+        S: Stream<Item = Result<T, E>> + Send + Sync + 'static + Unpin,
+        T: Send + Sync + 'static,
+        E: Send + Sync + 'static,
+        P::Msg: Sync,
+    {
+        StreamBuilder::new(f, self)
+    }
+
     async fn handle_err(
         &mut self,
         e: Box<dyn Any + Send>,