Implement the ability to manually work with channels from an actor

commit f42d7e9
Author: EnvOut <[email protected]>
Date:   Fri Nov 29 17:53:45 2024 +0200

    Implement the ability to manually work with channels from an actor

commit b0ee60e
Author: EnvOut <[email protected]>
Date:   Fri Nov 29 13:42:31 2024 +0200

    global system

commit aa381d8
Author: EnvOut <[email protected]>
Date:   Sun Nov 24 23:47:28 2024 +0200

    cargo fmt

commit 75746fc
Author: EnvOut <[email protected]>
Date:   Sun Nov 24 23:42:59 2024 +0200

    timeout decorator

Author: EnvOut

Cargo.lock

@@ -17,13 +17,11 @@ thiserror = { workspace = true }
 tracing = { workspace = true }
 paste = "1.0"
 derive_more = { workspace = true }
-derive-new = { workspace = true }
 bytes = { version = "1", optional = true }
-rand = "0.8"
-
-either = "1.13.0"
+lazy_static = "1.5.0"
 
 [dev-dependencies]
+more-asserts = "0.3.1"
 tracing-subscriber = { version = "0.3" }
 anyhow = "1"
 time = "0.3"

src/uactor/Cargo.toml

@@ -58,6 +58,7 @@ mod actor1 {
 
     impl Actor for Actor1 {
         type Context = Context;
+        type RouteMessage = Actor1Msg;
         type Inject = Services;
         type State = ();
     }
@@ -86,7 +87,7 @@ mod actor1 {
             Services { actor2_ref, .. }: &mut Self::Inject,
             msg: MessageWithoutReply,
             _ctx: &mut Context,
-            state: &Self::State,
+            _state: &Self::State,
         ) -> HandleResult {
             println!("actor1: Received {msg:?} message, sending PrintMessage to the actor2");
             actor2_ref.send(PrintMessage::new(msg.into()))?;
@@ -124,6 +125,7 @@ mod actor2 {
 
     impl Actor for Actor2 {
         type Context = Context;
+        type RouteMessage = Actor2Msg;
         type Inject = Services;
         type State = ();
     }
@@ -196,15 +198,16 @@ async fn main() -> anyhow::Result<()> {
 
     // Init system and register services
     let mut system = System::global()
-        .extension(service1)
-        .extension(service2)
-        .build();
+        .with(|mut system| {
+            system.extension(service1.clone())
+                .extension(service2.clone());
+        }).await;
 
     // Init actor (instance + spawn actor)
-    let (actor1_ref, actor1_stream) = system.register_ref::<Actor1, _, Actor1MpscRef>("actor1");
+    let (actor1_ref, actor1_stream) = system.register_ref::<Actor1, _, Actor1MpscRef>("actor1").await;
 
     // Init actor2 (instance + spawn actor)
-    let (actor2_ref, actor2_stream) = system.register_ref::<Actor2, _, Actor2MpscRef>("actor2");
+    let (actor2_ref, actor2_stream) = system.register_ref::<Actor2, _, Actor2MpscRef>("actor2").await;
 
     // Run actors
     let actor1 = Actor1;
@@ -214,17 +217,16 @@ async fn main() -> anyhow::Result<()> {
     system.spawn_actor(actor2_ref.name(), actor2, *actor2_ref.state(), (actor2_stream)).await?;
 
     // Case #1: send messages and call injected (not from &self) services inside handlers
-    println!(
-        "-- Case #1: send messages and call injected (not from &self) services inside handlers"
-    );
+    println!("-- Case #1: send messages and call injected (not from &self) services inside handlers");
     let pong1 = actor1_ref.ask::<PongMsg>(PingMsg).await?;
     let pong2 = actor2_ref.ask::<PongMsg>(PingMsg).await?;
-    println!("main: received {pong1:?} and {pong2:?} messages");
+    println!("main: received {pong1:?} and {pong2:?} messages\n");
 
     // Case #2: send message#1 to actor1 and reply to actor2 without actor2 reference inside message#1
-    println!("\n-- Case #2: send message#1 to actor1 and reply to actor2 without actor2 reference inside message#1");
+    println!("-- Case #2: send message#1 to actor1 and reply to actor2 without actor2 reference inside message#1");
     actor1_ref.send(MessageWithoutReply("login:password".to_owned()))?;
 
+    // wait for actors to finish
     tokio::time::sleep(1.std_milliseconds()).await;
     Ok(())
 }

src/uactor/examples/dependency_injection.rs

@@ -1,31 +1,34 @@
 use time::ext::NumericalStdDuration;
-use tokio::sync::mpsc::UnboundedSender;
+use tokio_stream::StreamExt;
 use uactor::actor::abstract_actor::MessageSender;
 
 use uactor::system::System;
 
-use crate::actor1::{Actor1, Actor1MpscRef};
 use crate::actor1::Actor1Msg;
-use crate::actor1::Actor1Ref;
-use crate::messages::{PingMsg, PongMsg};
+use crate::actor1::{Actor1, Actor1MpscRef};
+use crate::messages::{AskTicksCountMsg, TicksCount, UpdateMetrics};
+use more_asserts as ma;
+use uactor::actor::message::IntervalMessage;
+use uactor::data::datasource_decorator::DataSourceMapExt;
 
 mod messages {
     use uactor::actor::message::{Message, Reply};
 
-    pub struct PingMsg(pub Reply<PongMsg>);
+    pub struct AskTicksCountMsg(pub Reply<TicksCount>);
+    pub struct UpdateMetrics;
 
     #[derive(Debug)]
-    pub struct PongMsg;
+    pub struct TicksCount(pub usize);
 
-    uactor::message_impl!(PingMsg, PongMsg);
+    uactor::message_impl!(AskTicksCountMsg, TicksCount, UpdateMetrics);
 }
 
 mod actor1 {
     use uactor::actor::abstract_actor::{Actor, HandleResult, Handler};
-    use uactor::actor::context::Context;
+    use uactor::actor::context::{ActorContext, Context};
     use uactor::actor::message::IntervalMessage;
 
-    use crate::messages::{PingMsg, PongMsg};
+    use crate::messages::{AskTicksCountMsg, TicksCount, UpdateMetrics};
 
     #[derive(Default)]
     pub struct Actor1 {
@@ -34,21 +37,39 @@ mod actor1 {
 
     impl Actor for Actor1 {
         type Context = Context;
+        type RouteMessage = Actor1Msg;
         type Inject = ();
         type State = ();
     }
 
-    impl Handler<PingMsg> for Actor1 {
+    impl Handler<AskTicksCountMsg> for Actor1 {
         async fn handle(
             &mut self,
             _: &mut Self::Inject,
-            ping: PingMsg,
-            _ctx: &mut Context,
-            state: &Self::State,
+            AskTicksCountMsg(reply): AskTicksCountMsg,
+            ctx: &mut Context,
+            _state: &Self::State,
         ) -> HandleResult {
             println!("actor1: Received ping message");
-            let PingMsg(reply) = ping;
-            let _ = reply.send(PongMsg);
+            let _ = reply.send(TicksCount(self.interval_count as usize));
+            ctx.kill();
+            Ok(())
+        }
+    }
+
+    impl Handler<UpdateMetrics> for Actor1 {
+        async fn handle(
+            &mut self,
+            _: &mut Self::Inject,
+            _: UpdateMetrics,
+            _ctx: &mut Context,
+            _state: &Self::State,
+        ) -> HandleResult {
+            self.interval_count += 1;
+            println!(
+                "actor1: received {}nd UpdateMetrics message",
+                self.interval_count
+            );
             Ok(())
         }
     }
@@ -62,36 +83,41 @@ mod actor1 {
                 duration: _,
             }: IntervalMessage,
             _ctx: &mut Context,
-            state: &Self::State,
+            _state: &Self::State,
         ) -> HandleResult {
-            self.interval_count += 1;
-            println!(
-                "actor1: received {}nd interval message",
-                self.interval_count
-            );
+            // some important logic
             Ok(())
         }
     }
 
-    uactor::generate_actor_ref!(Actor1, { PingMsg });
+    uactor::generate_actor_ref!(Actor1, { AskTicksCountMsg, UpdateMetrics, IntervalMessage });
 }
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let mut system = System::global().build();
+    use uactor::data::datasource_decorator::DataSourceMapExt;
+    let mut system = System::global();
 
-    // 1 second interval
-    let interval = tokio::time::interval(1.std_seconds());
+    // 100 milliseconds interval for updating metrics
+    let metrics_update_interval = tokio::time::interval(100.std_milliseconds())
+        .map(|_: IntervalMessage| Actor1Msg::UpdateMetrics(UpdateMetrics));
+    // 1 second interval for other tasks
+    let other_interval = tokio::time::interval(1.std_seconds())
+        .map(Actor1Msg::IntervalMessage);
 
-    let (actor1_ref, actor1_stream) = system.register_ref::<Actor1, _, Actor1MpscRef>("actor1");
+    // Initialize actor's reference
+    let (actor1_ref, actor1_stream) = system.register_ref::<Actor1, _, Actor1MpscRef>("actor1").await;
 
+    // Spawn actor
     let actor1 = Actor1::default();
-    system.spawn_actor(actor1_ref.name(), actor1, *actor1_ref.state(), (actor1_stream, interval)).await?;
+    let (_state, actor_handle) = system.spawn_actor(actor1_ref.name(), actor1, *actor1_ref.state(), (actor1_stream, metrics_update_interval, other_interval)).await?;
 
-    let pong = actor1_ref.ask::<PongMsg>(PingMsg).await?;
-    println!("main: received {pong:?} message");
+    // Wait for 5 ticks
+    tokio::time::sleep(500.std_milliseconds()).await;
+    let TicksCount(ticks_count) = actor1_ref.ask::<TicksCount>(AskTicksCountMsg).await?;
+    ma::assert_ge!(ticks_count, 5, "waiting 5 ticks and expecting at least 5 messages received");
 
-    // waiting 10 seconds and expecting new message each 1 second
-    tokio::time::sleep(10.std_seconds()).await;
+    tokio::time::sleep(1.std_microseconds()).await;
+    assert!(actor_handle.is_finished(), "actor should be finished after receiving AskTicksCountMsg");
     Ok(())
 }