Repository: slawlor/ractor
Pull request #29 (2023-01-19 01:31:18)

Adding support for RPC macros
This PR cleans up the recent addition of macros along with some cleaner matching syntax

1. `call!` - Make an Rpc call without a timeout
2. `call_t!` - Make an Rpc call with a timeout
3. `forward!` - Make an Rpc call and forward it to another actor
4. `cast!` - Cast a message to an actor

There's more to come, but it cleans up the need to build a closure on the fly


Patch:
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index 237ffb0..e3fb30a 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ractor"
-version = "0.4.0"
+version = "0.4.1"
 authors = ["Sean Lawlor", "Evan Au", "Dillon George"]
 description = "A actor framework for Rust"
 documentation = "https://docs.rs/ractor"
diff --git a/ractor/examples/counter.rs b/ractor/examples/counter.rs
index c38eff4..f98bcbb 100644
--- a/ractor/examples/counter.rs
+++ b/ractor/examples/counter.rs
@@ -14,8 +14,7 @@
 
 extern crate ractor;
 
-use ractor::{Actor, ActorRef, RpcReplyPort};
-use tokio::time::Duration;
+use ractor::{call_t, Actor, ActorRef, RpcReplyPort};
 
 struct Counter;
 
@@ -76,10 +75,7 @@ async fn main() {
             .send_message(CounterMessage::Decrement(5))
             .expect("Failed to send message");
 
-        let rpc_result = actor
-            .call(CounterMessage::Retrieve, Some(Duration::from_millis(10)))
-            .await
-            .expect("Failed to send RPC");
+        let rpc_result = call_t!(actor, CounterMessage::Retrieve, 10);
 
         println!(
             "Count is: {}",
diff --git a/ractor/examples/monte_carlo.rs b/ractor/examples/monte_carlo.rs
index 78af0f5..2761bce 100644
--- a/ractor/examples/monte_carlo.rs
+++ b/ractor/examples/monte_carlo.rs
@@ -17,7 +17,7 @@
 
 use std::collections::HashMap;
 
-use ractor::{Actor, ActorId, ActorRef};
+use ractor::{cast, Actor, ActorId, ActorRef};
 use rand::{thread_rng, Rng};
 
 // ================== Player Actor ================== //
@@ -77,14 +77,18 @@ impl Actor for Game {
                 false => state.funds -= state.wager as i64,
             }
             state.results_vec.push(state.funds);
-            let _ = myself.cast(message);
+            cast!(myself, message).expect("Failed to send message");
         } else {
             // Now that the game is finished, the results of the game need to be reported
             // to the `GameManager`.
-            let _ = message.cast(GameManagerMessage {
-                id: myself.get_id(),
-                results: state.results_vec.clone(),
-            });
+            cast!(
+                message,
+                GameManagerMessage {
+                    id: myself.get_id(),
+                    results: state.results_vec.clone(),
+                }
+            )
+            .expect("Failed to send message");
             // Because the `GameManager` is monitoring this actor we can stop this actor
             myself.stop(None);
         }
@@ -141,7 +145,7 @@ impl Actor for GameManager {
             let (actor, _) = Actor::spawn_linked(None, Game, myself.clone().into())
                 .await
                 .expect("Failed to start game");
-            let _ = actor.cast(myself.clone());
+            cast!(actor, myself.clone()).expect("Failed to send message");
         }
 
         GameManagerState::new(self.num_games)
diff --git a/ractor/examples/philosophers.rs b/ractor/examples/philosophers.rs
index 8a2a5ca..c296593 100644
--- a/ractor/examples/philosophers.rs
+++ b/ractor/examples/philosophers.rs
@@ -20,7 +20,7 @@
 
 use std::collections::{HashMap, VecDeque};
 
-use ractor::{Actor, ActorId, ActorName, ActorRef, RpcReplyPort};
+use ractor::{cast, Actor, ActorId, ActorName, ActorRef, RpcReplyPort};
 use tokio::time::{Duration, Instant};
 
 // ============================ Fork Actor ============================ //
@@ -58,14 +58,14 @@ impl Fork {
                 match &state.owned_by {
                     Some(owner) => {
                         if !state.clean {
-                            let _ = owner.cast(PhilosopherMessage::GiveUpFork(myself.get_id()));
+                            let _ = cast!(owner, PhilosopherMessage::GiveUpFork(myself.get_id()));
                         }
                         // there's already an owner, backlog this message in priority
                         return Some(message);
                     }
                     None => {
                         // give the fork to the requester
-                        let _ = who.cast(PhilosopherMessage::ReceiveFork(myself.get_id()));
+                        let _ = cast!(who, PhilosopherMessage::ReceiveFork(myself.get_id()));
                         // set ownership
                         state.owned_by = Some(who.clone());
                     }
@@ -304,7 +304,7 @@ impl Actor for Philosopher {
     type State = PhilosopherState;
     async fn pre_start(&self, myself: ActorRef<Self>) -> Self::State {
         // initialize the simulation by making the philosopher's hungry
-        let _ = myself.cast(Self::Msg::BecomeHungry(0));
+        let _ = cast!(myself, Self::Msg::BecomeHungry(0));
         Self::State::new(self.left.clone(), self.right.clone())
     }
 
@@ -480,14 +480,8 @@ async fn main() {
     tokio::time::sleep(run_time).await;
     // collect the metrics from the philosophers, and they'll stop after reporting metrics
     for philosopher in philosophers.iter() {
-        let metrics = philosopher
-            .call(
-                PhilosopherMessage::SendMetrics,
-                Some(Duration::from_millis(50)),
-            )
-            .await
-            .expect("Failed to send message to philosopher")
-            .expect("Call result to philosopher wasn't successful");
+        let metrics = ractor::call_t!(philosopher, PhilosopherMessage::SendMetrics, 50)
+            .expect("Failed to perform RPC");
         results.insert(philosopher.get_name().unwrap(), Some(metrics));
     }
 
diff --git a/ractor/examples/ping_pong.rs b/ractor/examples/ping_pong.rs
index 22ec7d0..6a311c0 100644
--- a/ractor/examples/ping_pong.rs
+++ b/ractor/examples/ping_pong.rs
@@ -14,7 +14,7 @@
 
 extern crate ractor;
 
-use ractor::{Actor, ActorRef};
+use ractor::{cast, Actor, ActorRef};
 
 pub struct PingPong;
 
@@ -48,7 +48,7 @@ impl Actor for PingPong {
 
     async fn pre_start(&self, myself: ActorRef<Self>) -> Self::State {
         // startup the event processing
-        myself.send_message(Message::Ping).unwrap();
+        cast!(myself, Message::Ping).unwrap();
         // create the initial state
         0u8
     }
@@ -56,7 +56,7 @@ impl Actor for PingPong {
     async fn handle(&self, myself: ActorRef<Self>, message: Self::Msg, state: &mut Self::State) {
         if *state < 10u8 {
             message.print();
-            myself.send_message(message.next()).unwrap();
+            cast!(myself, message.next()).unwrap();
             *state += 1;
         } else {
             println!();
diff --git a/ractor/src/macros.rs b/ractor/src/macros.rs
index 306150a..e9cf424 100644
--- a/ractor/src/macros.rs
+++ b/ractor/src/macros.rs
@@ -14,13 +14,55 @@ macro_rules! cast {
     };
 }
 
-/// `call!`: Perform an inifinite-time remote procedure call to an [crate::Actor]
+/// `call!`: Perform an infinite-time remote procedure call to an [crate::Actor]
 ///
 /// * `$actor` - The actor to call
 /// * `$msg` - The message builder which takes in a [crate::port::RpcReplyPort] and emits a message which
 /// the actor supports
+/// * `$args` - (optional) Variable length arguments which will PRECEED the reply channel when
+/// constructing the message payload
 ///
 /// Returns [Ok(_)] with the result on successful RPC or [Err(crate::RactorErr)] on failure
+/// Example usage
+/// ```rust
+/// use ractor::{call, Actor, RpcReplyPort, ActorRef};
+/// struct TestActor;
+/// enum MessageFormat {
+///     TestRpc(String, RpcReplyPort<String>),
+/// }
+///
+/// #[async_trait::async_trait]
+/// impl Actor for TestActor {
+///     type Msg = MessageFormat;
+///
+///     type State = ();
+///
+///     async fn pre_start(&self, _this_actor: ActorRef<Self>) -> Self::State {}
+///
+///     async fn handle(
+///         &self,
+///         _this_actor: ActorRef<Self>,
+///         message: Self::Msg,
+///         _state: &mut Self::State,
+///     ) {
+///         match message {
+///             Self::Msg::TestRpc(arg, reply) => {
+///                 // An error sending means no one is listening anymore (the receiver was dropped),
+///                 // so we should shortcut the processing of this message probably!
+///                 if !reply.is_closed() {
+///                     let _ = reply.send(arg);
+///                 }
+///             }
+///         }
+///     }
+/// }
+///
+/// async fn test() {
+///     let (actor, _handle) = Actor::spawn(None, TestActor).await.unwrap();
+///     let result = call!(actor, MessageFormat::TestRpc, "Something".to_string()).unwrap();
+///     assert_eq!(result, "Something".to_string())
+/// }
+/// ```
 #[macro_export]
 macro_rules! call {
     ($actor:ident, $msg:expr) => {{
@@ -34,21 +76,85 @@ macro_rules! call {
             Err(e) => Err(e),
         }
     }};
+    ($actor:ident, $msg:expr, $($args:expr),*) => {{
+        let err = $actor
+            .call(|tx| $msg($($args),*, tx), None)
+            .await
+            .map_err($crate::RactorErr::from);
+        match err {
+            Ok($crate::rpc::CallResult::Success(ok_value)) => Ok(ok_value),
+            Ok(cr) => Err($crate::RactorErr::from(cr)),
+            Err(e) => Err(e),
+        }
+    }};
 }
 
 /// `call_t!`: Perform an finite-time remote procedure call to an [crate::Actor]
 ///
 /// * `$actor` - The actor to call
-/// * `$msg` - The message builder which takes in a [crate::port::RpcReplyPort] and emits a message which
-/// the actor supports
-/// * `$timeout` - The [tokio::time::Duration] timeout for how long the call can take before timing out
+/// * `$msg` - The message builder variant
+/// * `$timeout_ms` - the timeout in milliseconds for the remote procedure call
+/// * `$args` - (optional) Variable length arguments which will PRECEED the reply channel when
+/// constructing the message payload
 ///
 /// Returns [Ok(_)] with the result on successful RPC or [Err(crate::RactorErr)] on failure
+///
+/// Example usage
+/// ```rust
+/// use ractor::{call_t, Actor, ActorRef, RpcReplyPort};
+/// struct TestActor;
+/// enum MessageFormat {
+///     TestRpc(String, RpcReplyPort<String>),
+/// }
+///
+/// #[async_trait::async_trait]
+/// impl Actor for TestActor {
+///     type Msg = MessageFormat;
+///
+///     type State = ();
+///
+///     async fn pre_start(&self, _this_actor: ActorRef<Self>) -> Self::State {}
+///
+///     async fn handle(
+///         &self,
+///         _this_actor: ActorRef<Self>,
+///         message: Self::Msg,
+///         _state: &mut Self::State,
+///     ) {
+///         match message {
+///             Self::Msg::TestRpc(arg, reply) => {
+///                 // An error sending means no one is listening anymore (the receiver was dropped),
+///                 // so we should shortcut the processing of this message probably!
+///                 if !reply.is_closed() {
+///                     let _ = reply.send(arg);
+///                 }
+///             }
+///         }
+///     }
+/// }
+///
+/// async fn test() {
+///     let (actor, _handle) = Actor::spawn(None, TestActor).await.unwrap();
+///     let result = call_t!(actor, MessageFormat::TestRpc, 50, "Something".to_string()).unwrap();
+///     assert_eq!(result, "Something".to_string())
+/// }
+/// ```
 #[macro_export]
 macro_rules! call_t {
-    ($actor:ident, $msg:expr, $timeout:expr) => {{
+    ($actor:ident, $msg:expr, $timeout_ms:literal) => {{
+        let err = $actor
+            .call(|tx| $msg(tx), Some(tokio::time::Duration::from_millis($timeout_ms)))
+            .await
+            .map_err($crate::RactorErr::from);
+        match err {
+            Ok($crate::rpc::CallResult::Success(ok_value)) => Ok(ok_value),
+            Ok(cr) => Err($crate::RactorErr::from(cr)),
+            Err(e) => Err(e),
+        }
+    }};
+    ($actor:ident, $msg:expr, $timeout_ms:literal, $($args:expr),*) => {{
         let err = $actor
-            .call(|tx| $msg(tx), Some($timeout))
+            .call(|tx| $msg($($args),*, tx), Some(tokio::time::Duration::from_millis($timeout_ms)))
             .await
             .map_err($crate::RactorErr::from);
         match err {
@@ -59,7 +165,7 @@ macro_rules! call_t {
     }};
 }
 
-/// `forward!`: Perform an infinite-time remote procedure call to a [crate::Actor]
+/// `forward!`: Perform a remote procedure call to a [crate::Actor]
 /// and forwards the result to another actor if successful
 ///
 /// * `$actor` - The actors to call
@@ -67,6 +173,7 @@ macro_rules! call_t {
 /// the actor supports.
 /// * `$forward` - The [crate::ActorRef] to forward the call to
 /// * `$forward_mapping` - The message transformer from the RPC result to the forwarding actor's message format
+/// * `$timeout` - The [tokio::time::Duration] to allow the call to complete before timing out.
 ///
 /// Returns [Ok(())] on successful call forwarding, [Err(crate::RactorErr)] otherwies
 #[macro_export]
@@ -90,21 +197,6 @@ macro_rules! forward {
             Err(e) => Err(e),
         }
     }};
-}
-
-/// `forward_t!`: Perform an finite-time remote procedure call to a [crate::Actor]
-/// and forwards the result to another actor if successful
-///
-/// * `$actor` - The actors to call
-/// * `$msg` - The message builder, which takes in a [crate::port::RpcReplyPort] and emits a message which
-/// the actor supports.
-/// * `$forward` - The [crate::ActorRef] to forward the call to
-/// * `$forward_mapping` - The message transformer from the RPC result to the forwarding actor's message format
-/// * `$timeout` - The [tokio::time::Duration] to allow the call to complete before timing out.
-///
-/// Returns [Ok(())] on successful call forwarding, [Err(crate::RactorErr)] otherwies
-#[macro_export]
-macro_rules! forward_t {
     ($actor:ident, $msg:expr, $forward:ident, $forward_mapping:expr, $timeout:expr) => {{
         let future_or_err = $actor
             .call_and_forward(|tx| $msg(tx), &$forward, $forward_mapping, Some($timeout))
@@ -125,5 +217,3 @@ macro_rules! forward_t {
         }
     }};
 }
-
-// TODO: subscribe to output port?


Test patch:
diff --git a/ractor/src/rpc/tests.rs b/ractor/src/rpc/tests.rs
index 4c7f344..a02e47c 100644
--- a/ractor/src/rpc/tests.rs
+++ b/ractor/src/rpc/tests.rs
@@ -11,7 +11,7 @@ use std::sync::Arc;
 use tokio::time::Duration;
 
 use crate::rpc;
-use crate::{call, call_t, cast, forward, forward_t, Actor, ActorRef};
+use crate::{call, call_t, cast, forward, Actor, ActorRef};
 
 #[tokio::test]
 async fn test_rpc_cast() {
@@ -66,10 +66,10 @@ async fn test_rpc_cast() {
 #[tokio::test]
 async fn test_rpc_call() {
     struct TestActor;
-
     enum MessageFormat {
-        TestRpc(rpc::RpcReplyPort<String>),
-        TestTimeout(rpc::RpcReplyPort<String>),
+        Rpc(rpc::RpcReplyPort<String>),
+        Timeout(rpc::RpcReplyPort<String>),
+        MultiArg(String, u32, rpc::RpcReplyPort<String>),
     }
 
     #[async_trait::async_trait]
@@ -87,17 +87,20 @@ async fn test_rpc_call() {
             _state: &mut Self::State,
         ) {
             match message {
-                Self::Msg::TestRpc(reply) => {
+                Self::Msg::Rpc(reply) => {
                     // An error sending means no one is listening anymore (the receiver was dropped),
                     // so we should shortcut the processing of this message probably!
                     if !reply.is_closed() {
                         let _ = reply.send("howdy".to_string());
                     }
                 }
-                Self::Msg::TestTimeout(reply) => {
+                Self::Msg::Timeout(reply) => {
                     tokio::time::sleep(Duration::from_millis(100)).await;
                     let _ = reply.send("howdy".to_string());
                 }
+                Self::Msg::MultiArg(message, count, reply) => {
+                    let _ = reply.send(format!("{}-{}", message, count));
+                }
             }
         }
     }
@@ -106,38 +109,42 @@ async fn test_rpc_call() {
         .await
         .expect("Failed to start test actor");
 
-    let rpc_result = call_t!(
-        actor_ref,
-        MessageFormat::TestRpc,
-        Duration::from_millis(100)
-    )
-    .unwrap();
+    let rpc_result = call_t!(actor_ref, MessageFormat::Rpc, 100).unwrap();
     assert_eq!("howdy".to_string(), rpc_result);
 
-    let rpc_result = call!(actor_ref, MessageFormat::TestRpc).unwrap();
+    let rpc_result = call!(actor_ref, MessageFormat::Rpc).unwrap();
     assert_eq!("howdy".to_string(), rpc_result);
 
     let rpc_result = actor_ref
-        .call(MessageFormat::TestRpc, Some(Duration::from_millis(100)))
+        .call(MessageFormat::Rpc, Some(Duration::from_millis(100)))
         .await
         .expect("Failed to send message to actor")
         .expect("RPC didn't succeed");
     assert_eq!("howdy".to_string(), rpc_result);
 
-    let rpc_timeout = call_t!(
-        actor_ref,
-        MessageFormat::TestTimeout,
-        Duration::from_millis(10)
-    );
+    let rpc_timeout = call_t!(actor_ref, MessageFormat::Timeout, 10);
     assert!(rpc_timeout.is_err());
     println!("RPC Error {:?}", rpc_timeout);
 
+    let rpc_value = call!(actor_ref, MessageFormat::MultiArg, "Msg".to_string(), 32).unwrap();
+    assert_eq!("Msg-32".to_string(), rpc_value);
+
+    let rpc_value = call_t!(
+        actor_ref,
+        MessageFormat::MultiArg,
+        100,
+        "Msg".to_string(),
+        32
+    )
+    .unwrap();
+    assert_eq!("Msg-32".to_string(), rpc_value);
+
     // cleanup
     actor_ref.stop(None);
 
     tokio::time::sleep(Duration::from_millis(200)).await;
 
-    let rpc_send_fail = call!(actor_ref, MessageFormat::TestRpc);
+    let rpc_send_fail = call!(actor_ref, MessageFormat::Rpc);
     assert!(rpc_send_fail.is_err());
     handle.await.expect("Actor stopped with err");
 }
@@ -243,7 +250,7 @@ async fn test_rpc_call_forwarding() {
     )
     .expect("Failed to foward message");
 
-    forward_t!(
+    forward!(
         worker_ref,
         WorkerMessage::TestRpc,
         forwarder_ref,
diff --git a/ractor/src/time/tests.rs b/ractor/src/time/tests.rs
index 29f26ae..b693b33 100644
--- a/ractor/src/time/tests.rs
+++ b/ractor/src/time/tests.rs
@@ -114,7 +114,7 @@ async fn test_send_after() {
     // kill the actor
     actor_ref.stop(None);
 
-    tokio::time::sleep(Duration::from_millis(20)).await;
+    tokio::time::sleep(Duration::from_millis(50)).await;
     // make sure the actor is dead + the interval handle doesn't send again
     assert!(send_after_handle.is_finished());
     assert!(actor_handle.is_finished());