docs, in_request macro and tweaks

Cargo.toml

@@ -11,25 +11,30 @@ keywords = ["actor", "compute", "graph", "pipeline"]
 license = "Apache-2.0"
 readme = "README.md"
 repository = "https://github.com/farm-ng/hollywood/"
-version = "0.6.0"
+version = "0.7.0"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 async-trait = "0.1"
+ctrlc = "3.4"
 drawille = "0.3"
 # hollywood intends to use only basic features of egui, hence 
 # future versions of egui will likely/hopefully work
 eframe = {version = ">= 0.27, <1.0", features = ["wgpu"], optional = true}
+#version = ">= 0.27, <1.0", 
 env_logger = {version = "0.11", optional = true}
 grid = "0.13"
-hollywood_macros = {version = "0.6.0", path = "hollywood_macros"}
+hollywood_macros = {version = "0.7.0", path = "hollywood_macros"}
+linear_type = {version = "0.4"}
 # hollywood intends to use only very basic features of nalgebra, hence 
 # future versions of nalgebra before the major < 1.0 release are likely to work
 nalgebra = ">= 0.32, <1.0"
 petgraph = "0.6"
 rand = "0.8"
 rand_distr = "0.4"
+tracing = "0.1"
+tracing-subscriber = "0.3"
 # executor feature needed
 tokio = {version = "1.28", features = ["full"]}
 tokio-stream = "0.1"

README.md

@@ -1,5 +1,96 @@
 # hollywood
 
-Hollywood is an actor framework, with focus on representing actors with heterogeneous
-inputs and outputs which are arranged in a non-cyclic compute graph/pipeline. The design
-intend is simplicity and minimal boilerplate code.
+Hollywood is an actor framework for Rust -  with focus on representing actors with
+heterogeneous inputs and outputs which are arranged in a non-cyclic compute graph/pipeline. The
+design intend is simplicity and minimal boilerplate code. Hence, Hollywood is an abstraction over
+async rust in general and the asynchronous tokio runtime in particular. If you do not seek
+such an abstraction, you may want to use tokio (or another async runtime) directly.
+
+Actors are stateful entities that communicate with each other by sending messages. An actor
+actor receives streams of messages through its inbound channels, processes them and sends
+messages to other actors through its outbound channels. Actors are either stateless so that
+its outbound streams are a pure function of its inbound streams, or have an internal state which
+is updated by incoming messages and may influences the content of its outbound messages.
+
+Actors are arranged in a compute pipeline (or a directed acyclic graph to be more precise). The
+first layer of the graph consists of a set of one or more source actors whose outbound streams
+are fed by an external resource. A typical example of a source actors, include sensor drivers,
+or log file readers. Terminal notes in the pipeline are actors which have either no outbound
+channels or which outbound channels are not connected. Terminal actors are often sink notes
+which feed data to an external resource. Example of a sink actor are robot manipulators,
+log file writer or visualization components.
+
+In addition to the feed-forward connections between actors, actors can also communicate with
+each other through a set of request-reply channels. There is no restriction on which actor pairs
+can be connected with such request-reply channels. For example, a request-reply channel can
+be use to create a feedback loop in the compute flow.
+
+## Example
+
+The following example demonstrates how to create a simple pipeline with a periodic source actor
+which feeds a moving average actor. The moving average actor is connected to two printer actors
+which print the time stamp and the moving average to the console.
+
+```rust
+use hollywood::actors::{Periodic, Printer, PrinterProp};
+use hollywood::prelude::*;
+use hollywood::example_actors::moving_average::{MovingAverage, MovingAverageProp, MovingAverageState};
+
+let pipeline = Hollywood::configure(&mut |context| {
+    let mut timer = Periodic::new_with_period(context, 1.0);
+    let mut moving_average = MovingAverage::from_prop_and_state(
+        context,
+        MovingAverageProp {
+            alpha: 0.3,
+            timeout: 5.0,
+        },
+        MovingAverageState::default(),
+    );
+    let mut time_printer = Printer::<f64>::from_prop_and_state(
+        context,
+        PrinterProp {
+            topic: "time".to_string(),
+        },
+        NullState::default(),
+    );
+    let mut average_printer = Printer::<f64>::from_prop_and_state(
+        context,
+        PrinterProp {
+            topic: "average".to_string(),
+        },
+        NullState::default(),
+    );
+    timer
+        .outbound
+        .time_stamp
+        .connect(context, &mut moving_average.inbound.value);
+    timer
+        .outbound
+        .time_stamp
+        .connect(context, &mut time_printer.inbound.printable);
+    moving_average
+        .outbound
+        .average
+        .connect(context, &mut average_printer.inbound.printable);
+});
+pipeline.print_flow_graph();   
+pipeline.run();
+```
+
+The output of the `print_flow_graph` method is:
+
+```plaintext
+*Periodic_0*
+|   time_stamp   |
+        ⡏⠉⠑⠒⠢⠤⠤⣀⣀
+        ⡇        ⠉⠉⠑⠒⠢⠤⠤⣀⣀
+        ⠁                 ⠁
+|     Value      |                  |   Printable    |
+*MovingAverage_0*                  *Printer(time)_0*
+|    average     |
+        ⡇
+        ⡇
+        ⠁
+|   Printable    |
+*Printer(average)*  
+```

examples/egui.rs

@@ -1,10 +1,12 @@
-use std::time::Duration;
-
+use eframe::egui;
 use hollywood::actors::egui::EguiActor;
 use hollywood::actors::egui::EguiAppFromBuilder;
 use hollywood::actors::egui::GenericEguiBuilder;
 use hollywood::actors::egui::Stream;
 use hollywood::prelude::*;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::time::Duration;
 
 #[derive(Clone, Debug)]
 #[actor_inputs(
@@ -39,7 +41,7 @@ impl IsInRequestMessage for ContentGeneratorInRequestMessage {
 }
 
 impl HasOnRequestMessage for ContentGeneratorInRequestMessage {
-    fn on_message(
+    fn on_request(
         self,
         _prop: &Self::Prop,
         state: &mut Self::State,
@@ -49,7 +51,7 @@ impl HasOnRequestMessage for ContentGeneratorInRequestMessage {
         match self {
             ContentGeneratorInRequestMessage::Reset(msg) => {
                 state.offset = -state.last_x;
-                msg.reply(|_| state.last_x);
+                msg.reply(state.last_x);
             }
         }
     }
@@ -204,7 +206,8 @@ pub struct EguiAppExample {
         tokio::sync::mpsc::UnboundedReceiver<RequestWithReplyChannel<String, String>>,
     pub out_reply_recv: tokio::sync::mpsc::UnboundedReceiver<ReplyMessage<f64>>,
     pub out_request_sender: tokio::sync::mpsc::UnboundedSender<()>,
-    pub cancel_request_sender: tokio::sync::mpsc::UnboundedSender<hollywood::CancelRequest>,
+    pub cancel_request_sender: tokio::sync::mpsc::UnboundedSender<CancelRequest>,
+    pub on_exit_recv: Arc<Mutex<tokio::sync::mpsc::UnboundedReceiver<()>>>,
 
     pub x: f64,
     pub sin_value: f64,
@@ -219,6 +222,7 @@ impl EguiAppFromBuilder<EguiAppExampleBuilder> for EguiAppExample {
             in_request_recv: builder.in_request_from_actor_recv,
             out_request_sender: builder.out_request_to_actor_sender,
             cancel_request_sender: builder.cancel_request_sender.unwrap(),
+            on_exit_recv: builder.on_exit_recv,
             x: 0.0,
             sin_value: 0.0,
             cos_value: 0.0,
@@ -229,12 +233,14 @@ impl EguiAppFromBuilder<EguiAppExampleBuilder> for EguiAppExample {
 
     type State = String;
 }
-use eframe::egui;
-use hollywood::OutRequestChannel;
-use hollywood::ReplyMessage;
-use hollywood::RequestWithReplyChannel;
+
 impl eframe::App for EguiAppExample {
     fn update(&mut self, ctx: &egui::Context, _frame: &mut eframe::Frame) {
+        // Close the window if the on exit signal from the pipeline is received
+        if self.on_exit_recv.lock().unwrap().try_recv().is_ok() {
+            ctx.send_viewport_cmd(egui::ViewportCommand::Close);
+        }
+        // Handle incoming messages
         while let Ok(value) = self.message_recv.try_recv() {
             match value.msg {
                 PlotMessage::SinPlot((x, y)) => {
@@ -247,14 +253,14 @@ impl eframe::App for EguiAppExample {
                 }
             }
         }
+        // Handle incoming requests
+        while let Ok(value) = self.in_request_recv.try_recv() {
+            value.reply("reply".to_owned());
+        }
+        // Process replies (to reset request) from the content generator
         while let Ok(value) = self.out_reply_recv.try_recv() {
             println!("Reply: {:?}", value);
         }
-        while let Ok(value) = self.in_request_recv.try_recv() {
-            //println!("Request: {:?}", value);
-
-            value.reply(|_| "reply".to_owned());
-        }
 
         egui::CentralPanel::default().show(ctx, |ui| {
             ui.heading("Hello, egui!");
@@ -263,6 +269,7 @@ impl eframe::App for EguiAppExample {
             ui.label(format!("cos(y): {}", self.cos_value));
 
             if ui.button("Reset").clicked() {
+                // send a reset request to the content generator
                 self.out_request_sender.send(()).unwrap();
             }
         });
@@ -271,9 +278,7 @@ impl eframe::App for EguiAppExample {
     }
 
     fn on_exit(&mut self, _gl: Option<&eframe::glow::Context>) {
-        self.cancel_request_sender
-            .send(hollywood::CancelRequest::Cancel(()))
-            .unwrap();
+        self.cancel_request_sender.send(CancelRequest).unwrap();
     }
 }
 
@@ -344,7 +349,6 @@ pub fn run_egui_app_on_man_thread<
 >(
     builder: Builder,
 ) {
-    env_logger::init();
     let options = eframe::NativeOptions {
         viewport: egui::ViewportBuilder::default().with_inner_size([640.0, 480.0]),
         renderer: eframe::Renderer::Wgpu,
@@ -363,6 +367,8 @@ pub fn run_egui_app_on_man_thread<
 }
 
 fn main() {
+    tracing_subscriber::fmt::init();
+
     tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .build()

examples/moving_average.rs

@@ -56,6 +56,8 @@ pub async fn run_moving_average_example() {
 }
 
 fn main() {
+    tracing_subscriber::fmt::init();
+
     tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .build()

examples/nudge.rs

@@ -24,6 +24,8 @@ pub async fn run_tick_print_example() {
 }
 
 fn main() {
+    tracing_subscriber::fmt::init();
+
     tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .build()

examples/one_dim_robot.rs

@@ -111,6 +111,8 @@ async fn run_robot_example() {
 }
 
 fn main() {
+    tracing_subscriber::fmt::init();
+
     tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .build()

examples/print_ticks.rs

@@ -26,6 +26,8 @@ pub async fn run_tick_print_example() {
 }
 
 fn main() {
+    tracing_subscriber::fmt::init();
+
     tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .build()

examples/zip.rs

@@ -51,6 +51,8 @@ pub async fn run_tick_print_example() {
 }
 
 fn main() {
+    tracing_subscriber::fmt::init();
+
     tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .build()

hollywood_macros/Cargo.toml

@@ -11,7 +11,7 @@ keywords = ["actor", "compute", "graph", "pipeline"]
 license = "Apache-2.0"
 readme = "../README.md"
 repository = "https://github.com/farm-ng/hollywood/tree/main/hollywood_macros"
-version = "0.6.0"
+version = "0.7.0"
 
 [lib]
 proc-macro = true