Repository: slawlor/ractor
Pull request #5 (2023-01-05 16:51:41)

Some additional cleanups + initial workings of an RPC workflow
Adding an RPC workflow and cleanup of how message posting is handled.

1. Removed the ability to send a `BoxedMessage` directly on the actor, such that you always have to send a strong-typed message to an actor implementation.
2. Wrapped a send error into a cleaner type noting that an actor or at least a channel is dead when sending


Patch:
diff --git a/.github/workflows/publish.yaml b/.github/workflows/publish.yaml
new file mode 100644
index 0000000..ad2c5de
--- /dev/null
+++ b/.github/workflows/publish.yaml
@@ -0,0 +1,28 @@
+name: Publish
+
+on:
+  release:
+    types: [published]
+
+jobs:
+  publish:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ubuntu-latest]
+
+    steps:
+    - uses: hecrj/setup-rust-action@v1
+      with:
+        rust-version: ${{ matrix.rust }}
+    - uses: actions/checkout@main
+    - name: Login to crates.io
+      run: cargo login $CRATES_IO_TOKEN
+      env:
+        CRATES_IO_TOKEN: ${{ secrets.crates_io_token }}
+    - name: Dry run publish ractor
+      run: cargo publish --dry-run --manifest-path Cargo.toml -p ractor
+    - name: Publish crate ractor
+      run: cargo publish --manifest-path Cargo.toml -p ractor
+      env:
+        CARGO_REGISTRY_TOKEN: ${{ secrets.crates_io_token }}
\ No newline at end of file
diff --git a/ractor-repl/src/main.rs b/ractor-repl/src/main.rs
index d7955ca..5bc3963 100644
--- a/ractor-repl/src/main.rs
+++ b/ractor-repl/src/main.rs
@@ -9,7 +9,7 @@ mod ping_pong;
 #[tokio::main]
 async fn main() {
     let actor_handler = ping_pong::PingPong;
-    let (agent, ports) = ractor::Actor::new(None, actor_handler);
-    let (_actor_ref, actor_handle) = agent.start(ports, None).await.unwrap();
+    let (the_actor, ports) = ractor::Actor::new(None, actor_handler);
+    let (_actor_ref, actor_handle) = the_actor.start(ports, None).await.unwrap();
     actor_handle.await.unwrap();
 }
diff --git a/ractor-repl/src/ping_pong.rs b/ractor-repl/src/ping_pong.rs
index fb56a58..6347af9 100644
--- a/ractor-repl/src/ping_pong.rs
+++ b/ractor-repl/src/ping_pong.rs
@@ -36,7 +36,7 @@ impl ActorHandler for PingPong {
     async fn pre_start(&self, myself: ActorCell) -> Self::State {
         println!("pre_start called");
         // startup the event processing
-        let _ = myself.send_message_t(Message::Ping);
+        self.send_message(myself, Message::Ping).unwrap();
         // create the initial state
         0u8
     }
@@ -51,8 +51,9 @@ impl ActorHandler for PingPong {
     }
 
     /// Invoked after an actor has been stopped.
-    async fn post_stop(&self, _this_actor: ActorCell, _state: &Self::State) {
+    async fn post_stop(&self, _this_actor: ActorCell, _state: Self::State) -> Self::State {
         println!("post_stop called");
+        _state
     }
 
     async fn handle(
@@ -72,16 +73,16 @@ impl ActorHandler for PingPong {
             }
             Message::Stop => {
                 println!("Stopping");
-                let _ = myself.stop().await;
+                myself.stop();
                 return None;
             }
         };
         if *state > 10u8 {
             println!("\nSending stop...");
-            let _ = myself.send_message_t(Message::Stop);
+            self.send_message(myself, Message::Stop).unwrap();
         } else {
             let next = message.next();
-            let _ = myself.send_message_t(next);
+            self.send_message(myself, next).unwrap();
         }
         result
     }
diff --git a/ractor/Cargo.toml b/ractor/Cargo.toml
index 570117b..4f2045d 100644
--- a/ractor/Cargo.toml
+++ b/ractor/Cargo.toml
@@ -10,6 +10,6 @@ repository = "https://github.com/slawlor/ractor"
 readme = "../README.md"
 
 [dependencies]
-async-recursion = "1"
 async-trait = "0.1"
+dashmap = "5"
 tokio = { version = "1.23", features = ["rt", "time", "sync", "macros"]}
\ No newline at end of file
diff --git a/ractor/src/actor/actor_cell.rs b/ractor/src/actor/actor_cell.rs
index 87a7d6d..f6a4ade 100644
--- a/ractor/src/actor/actor_cell.rs
+++ b/ractor/src/actor/actor_cell.rs
@@ -9,14 +9,19 @@ use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 
 use tokio::sync::mpsc;
+use tokio::time::Duration;
 
+use super::errors::MessagingErr;
 use super::messages::{BoxedMessage, Signal};
 use super::supervision::SupervisionTree;
 use super::SupervisionEvent;
-use crate::port::input::InputPortReceiver;
-use crate::ActorId;
+use crate::port::{
+    BoundedInputPort, BoundedInputPortReceiver, InputPort, InputPortReceiver, RpcReplyPort,
+};
+use crate::rpc::{self, CallResult};
+use crate::{ActorHandler, ActorId, Message};
 
-/// The status of the agent
+/// The status of the actor
 #[derive(Debug, Clone, Eq, PartialEq, Copy)]
 #[repr(u8)]
 pub enum ActorStatus {
@@ -34,10 +39,10 @@ pub enum ActorStatus {
     Stopped = 5u8,
 }
 
-/// The collection of ports an agent needs to listen to
+/// The collection of ports an actor needs to listen to
 pub struct ActorPortSet {
-    pub(crate) signal_rx: mpsc::Receiver<Signal>,
-    pub(crate) supervisor_rx: mpsc::UnboundedReceiver<SupervisionEvent>,
+    pub(crate) signal_rx: BoundedInputPortReceiver<Signal>,
+    pub(crate) supervisor_rx: InputPortReceiver<SupervisionEvent>,
     pub(crate) message_rx: InputPortReceiver<BoxedMessage>,
 }
 
@@ -46,9 +51,9 @@ struct ActorProperties {
     id: ActorId,
     name: Option<String>,
     status: Arc<AtomicU8>,
-    signal: mpsc::Sender<Signal>,
-    supervision: mpsc::UnboundedSender<SupervisionEvent>,
-    message: mpsc::UnboundedSender<BoxedMessage>,
+    signal: BoundedInputPort<Signal>,
+    supervision: InputPort<SupervisionEvent>,
+    message: InputPort<BoxedMessage>,
     tree: SupervisionTree,
 }
 impl ActorProperties {
@@ -56,16 +61,16 @@ impl ActorProperties {
         name: Option<String>,
     ) -> (
         Self,
-        mpsc::Receiver<Signal>,
-        mpsc::UnboundedReceiver<SupervisionEvent>,
-        mpsc::UnboundedReceiver<BoxedMessage>,
+        BoundedInputPortReceiver<Signal>,
+        InputPortReceiver<SupervisionEvent>,
+        InputPortReceiver<BoxedMessage>,
     ) {
         let (tx, rx) = mpsc::channel(2);
         let (tx2, rx2) = mpsc::unbounded_channel();
         let (tx3, rx3) = mpsc::unbounded_channel();
         (
             Self {
-                id: crate::AGENT_ID_ALLOCATOR.fetch_add(1u64, Ordering::Relaxed),
+                id: crate::ACTOR_ID_ALLOCATOR.fetch_add(1u64, Ordering::Relaxed),
                 name,
                 status: Arc::new(AtomicU8::new(ActorStatus::Unstarted as u8)),
                 signal: tx,
@@ -94,22 +99,16 @@ impl ActorProperties {
         self.status.store(status as u8, Ordering::Relaxed);
     }
 
-    pub async fn send_signal(&self, signal: Signal) -> Result<(), mpsc::error::SendError<Signal>> {
-        self.signal.send(signal).await
+    pub fn send_signal(&self, signal: Signal) -> Result<(), MessagingErr> {
+        self.signal.try_send(signal).map_err(|e| e.into())
     }
 
-    pub fn send_supervisor_evt(
-        &self,
-        message: SupervisionEvent,
-    ) -> Result<(), mpsc::error::SendError<SupervisionEvent>> {
-        self.supervision.send(message)
+    pub fn send_supervisor_evt(&self, message: SupervisionEvent) -> Result<(), MessagingErr> {
+        self.supervision.send(message).map_err(|e| e.into())
     }
 
-    pub fn send_message(
-        &self,
-        message: BoxedMessage,
-    ) -> Result<(), mpsc::error::SendError<BoxedMessage>> {
-        self.message.send(message)
+    pub fn send_message(&self, message: BoxedMessage) -> Result<(), MessagingErr> {
+        self.message.send(message).map_err(|e| e.into())
     }
 }
 
@@ -121,7 +120,11 @@ pub struct ActorCell {
 
 impl std::fmt::Debug for ActorCell {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "Agent {}", self.get_id())
+        if let Some(name) = self.get_name() {
+            write!(f, "Actor '{}' (id: {})", name, self.get_id())
+        } else {
+            write!(f, "Actor with id: {}", self.get_id())
+        }
     }
 }
 
@@ -141,12 +144,12 @@ impl ActorCell {
         )
     }
 
-    /// Retrieve the agent's id
+    /// Retrieve the actor's id
     pub fn get_id(&self) -> ActorId {
         self.inner.id
     }
 
-    /// Retrieve the agent's name
+    /// Retrieve the actor's name
     pub fn get_name(&self) -> Option<String> {
         self.inner.name.clone()
     }
@@ -161,82 +164,134 @@ impl ActorCell {
         self.inner.set_status(status)
     }
 
-    // /// Returns the state object, only in the event of a dead actor. Otherwise
-    // /// returns None. We cannot read the state of a live actor (for thread safety)
-    // pub fn get_state(&self) -> Option<Box<dyn std::any::Any>> {
-    //     None
-    // }
-
     /// Terminate yourself and all children beneath you
-    pub async fn terminate(&self) {
+    pub fn terminate(&self) {
         // we don't need to nofity of exit if we're already stopping or stopped
         if self.get_status() as u8 <= ActorStatus::Upgrading as u8 {
             // kill myself immediately. Ignores failures, as a failure means either
             // 1. we're already dead or
             // 2. the channel is full of "signals"
-            let _ = self.send_signal(Signal::Exit).await;
+            self.stop();
         }
 
         // notify children they should die. They will unlink themselves from the supervisor
-        self.inner.tree.terminate_children().await;
+        self.inner.tree.terminate_children();
     }
 
     /// Link another actor to the supervised list
-    pub async fn link(&self, other: ActorCell) {
-        other.inner.tree.insert_parent(self.clone()).await;
-        self.inner.tree.insert_child(other).await;
+    pub fn link(&self, other: ActorCell) {
+        other.inner.tree.insert_parent(self.clone());
+        self.inner.tree.insert_child(other);
     }
 
     /// Unlink another actor from the supervised list
-    pub async fn unlink(&self, other: ActorCell) {
-        other.inner.tree.remove_parent(self.clone()).await;
-        self.inner.tree.remove_child(other).await;
+    pub fn unlink(&self, other: ActorCell) {
+        other.inner.tree.remove_parent(self.clone());
+        self.inner.tree.remove_child(other);
+    }
+    /// Stop this actor, but sending Signal::Exit
+    pub fn stop(&self) {
+        // ignore failures, since either the actor is already dead
+        // or the channel is full of "signals" which is also fine
+        // since it'll die shortly
+        let _ = self.inner.send_signal(Signal::Exit);
     }
 
-    /// Send a signal event to the signal port
-    pub async fn send_signal(&self, signal: Signal) -> Result<(), mpsc::error::SendError<Signal>> {
-        self.inner.send_signal(signal).await
+    /// Send a supervisor event to the supervisory port
+    pub fn send_supervisor_evt(&self, message: SupervisionEvent) -> Result<(), MessagingErr> {
+        self.inner.send_supervisor_evt(message)
     }
 
-    /// Stop this agent, but sending Signal::Exit
-    pub async fn stop(&self) -> Result<(), mpsc::error::SendError<Signal>> {
-        self.send_signal(Signal::Exit).await
+    /// Send a strongly-typed message, constructing the boxed message on the fly
+    ///
+    /// Note: The type requirement of `TActor` assures that `TMsg` is the supported
+    /// message type for `TActor` such that we can't send boxed messages of an unsupported
+    /// type to the specified actor.
+    pub fn send_message<TActor, TMsg>(&self, message: TMsg) -> Result<(), MessagingErr>
+    where
+        TActor: ActorHandler<Msg = TMsg>,
+        TMsg: Message,
+    {
+        self.inner.send_message(BoxedMessage::new(message))
     }
 
-    /// Send a supervisor event to the supervisory port
-    pub fn send_supervisor_evt(
-        &self,
-        message: SupervisionEvent,
-    ) -> Result<(), mpsc::error::SendError<SupervisionEvent>> {
-        self.inner.send_supervisor_evt(message)
+    /// Notify the supervisors that a supervision event occurred
+    pub fn notify_supervisors<TActor, TState>(&self, evt: SupervisionEvent)
+    where
+        TActor: ActorHandler<State = TState>,
+        TState: crate::State,
+    {
+        self.inner.tree.notify_supervisors::<TActor, _>(evt)
     }
 
-    /// Send a message to the regular message handling port
-    pub fn send_message(
-        &self,
-        message: BoxedMessage,
-    ) -> Result<(), mpsc::error::SendError<BoxedMessage>> {
-        self.inner.send_message(message)
+    /// Alias of [rpc::cast]
+    pub fn cast<TActor, TMsg>(&self, msg: TMsg) -> Result<(), MessagingErr>
+    where
+        TActor: ActorHandler<Msg = TMsg>,
+        TMsg: Message,
+    {
+        rpc::cast::<TActor, _>(self, msg)
     }
 
-    /// Send a strongly-typed message, constructing the boxed message on the fly
-    pub fn send_message_t<TMsg: crate::Message>(
+    /// Alias of [rpc::call]
+    pub async fn call<TActor, TMsg, TReply, TMsgBuilder>(
         &self,
-        message: TMsg,
-    ) -> Result<(), mpsc::error::SendError<BoxedMessage>> {
-        self.inner.send_message(BoxedMessage::new(message, true))
+        msg_builder: TMsgBuilder,
+        timeout_option: Option<Duration>,
+    ) -> Result<CallResult<TReply>, MessagingErr>
+    where
+        TActor: ActorHandler<Msg = TMsg>,
+        TMsg: Message,
+        TMsgBuilder: FnOnce(RpcReplyPort<TReply>) -> TMsg,
+    {
+        rpc::call::<TActor, TMsg, TReply, TMsgBuilder>(self, msg_builder, timeout_option).await
     }
 
-    /// Notify the supervisors that a supervision event occurred
-    pub async fn notify_supervisors(
+    /// Alias of [rpc::call_and_forward]
+    pub fn call_and_forward<
+        TActor,
+        TForwardActor,
+        TMsg,
+        TReply,
+        TMsgBuilder,
+        FwdMapFn,
+        TForwardMessage,
+    >(
         &self,
-        message: SupervisionEvent,
-    ) -> Result<(), mpsc::error::SendError<SupervisionEvent>> {
-        self.inner.tree.notify_supervisors(message).await
+        msg_builder: TMsgBuilder,
+        response_forward: ActorCell,
+        forward_mapping: FwdMapFn,
+        timeout_option: Option<Duration>,
+    ) -> Result<tokio::task::JoinHandle<CallResult<Result<(), MessagingErr>>>, MessagingErr>
+    where
+        TActor: ActorHandler<Msg = TMsg>,
+        TMsg: Message,
+        TReply: Message,
+        TMsgBuilder: FnOnce(RpcReplyPort<TReply>) -> TMsg,
+        TForwardActor: ActorHandler<Msg = TForwardMessage>,
+        FwdMapFn: FnOnce(TReply) -> TForwardMessage + Send + 'static,
+        TForwardMessage: Message,
+    {
+        rpc::call_and_forward::<
+            TActor,
+            TForwardActor,
+            TMsg,
+            TReply,
+            TMsgBuilder,
+            FwdMapFn,
+            TForwardMessage,
+        >(
+            self,
+            msg_builder,
+            response_forward,
+            forward_mapping,
+            timeout_option,
+        )
     }
 
+    /// Test utility to retrieve a clone of the underlying supervision tree
     #[cfg(test)]
-    pub fn get_tree(&self) -> SupervisionTree {
+    pub(crate) fn get_tree(&self) -> SupervisionTree {
         self.inner.tree.clone()
     }
 }
diff --git a/ractor/src/actor/errors.rs b/ractor/src/actor/errors.rs
index 15294a2..d365265 100644
--- a/ractor/src/actor/errors.rs
+++ b/ractor/src/actor/errors.rs
@@ -14,7 +14,7 @@ pub enum SpawnErr {
     StartupPanic(String),
     /// Actor failed to startup because the startup task was cancelled
     StartupCancelled,
-    /// An agent cannot be started > 1 time
+    /// An actor cannot be started > 1 time
     ActorAlreadyStarted,
 }
 
@@ -27,11 +27,11 @@ impl Display for SpawnErr {
             Self::StartupCancelled => {
                 write!(
                     f,
-                    "Actor failed to startup due to cancelled processing task"
+                    "Actor failed to startup due to processing task being cancelled"
                 )
             }
             Self::ActorAlreadyStarted => {
-                write!(f, "Actor cannot be started more than once")
+                write!(f, "Actor cannot be re-started more than once")
             }
         }
     }
@@ -39,14 +39,14 @@ impl Display for SpawnErr {
 
 /// Actor processing loop errors
 #[derive(Debug)]
-pub enum ActorProcessingErr {
+pub enum ActorErr {
     /// Actor had a task cancelled internally during processing
     Cancelled,
     /// Actor had an internal panic
     Panic(String),
 }
 
-impl Display for ActorProcessingErr {
+impl Display for ActorErr {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::Panic(panic_msg) => {
@@ -58,3 +58,46 @@ impl Display for ActorProcessingErr {
         }
     }
 }
+
+/// A messaging error has occurred
+#[derive(Debug)]
+pub enum MessagingErr {
+    /// The channel you're trying to send a message too has been dropped/closed.
+    /// If you're sending to an [crate::ActorCell] then that means the actor has died
+    /// (failure or not).
+    ChannelClosed,
+}
+
+impl<T> From<tokio::sync::mpsc::error::SendError<T>> for MessagingErr {
+    fn from(_: tokio::sync::mpsc::error::SendError<T>) -> Self {
+        Self::ChannelClosed
+    }
+}
+
+impl<T> From<tokio::sync::broadcast::error::SendError<T>> for MessagingErr {
+    fn from(_: tokio::sync::broadcast::error::SendError<T>) -> Self {
+        Self::ChannelClosed
+    }
+}
+
+impl<T> From<tokio::sync::watch::error::SendError<T>> for MessagingErr {
+    fn from(_: tokio::sync::watch::error::SendError<T>) -> Self {
+        Self::ChannelClosed
+    }
+}
+
+impl<T> From<tokio::sync::mpsc::error::TrySendError<T>> for MessagingErr {
+    fn from(_: tokio::sync::mpsc::error::TrySendError<T>) -> Self {
+        Self::ChannelClosed
+    }
+}
+
+impl Display for MessagingErr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::ChannelClosed => {
+                write!(f, "Messaging failed because channel is closed")
+            }
+        }
+    }
+}
diff --git a/ractor/src/actor/messages.rs b/ractor/src/actor/messages.rs
index 225fc64..3993c1b 100644
--- a/ractor/src/actor/messages.rs
+++ b/ractor/src/actor/messages.rs
@@ -3,70 +3,87 @@
 // This source code is licensed under both the MIT license found in the
 // LICENSE-MIT file in the root directory of this source tree.
 
-//! Messages which are built-in for the agent
+//! Messages which are built-in for the actor
 
 use std::any::Any;
 use std::fmt::Debug;
 
-use crate::Message;
+use crate::{Message, State};
+
+/// An error downcasting a boxed item to a strong type
+#[derive(Debug)]
+pub struct BoxedDowncastErr;
 
 /// A "boxed" message denoting a strong-type message
 /// but generic so it can be passed around without type
 /// constraints
 pub struct BoxedMessage {
-    /// Flag denoting if message can be consumed 1 time
-    pub one_time: bool,
     /// The message value
     pub msg: Option<Box<dyn Any + Send>>,
 }
 
-/// An error downcasting a boxed message to a strong type
-pub struct DowncastBoxedMessageErr;
-
 impl BoxedMessage {
     /// Create a new [BoxedMessage] from a strongly-typed message
-    /// and a flag denoting if the message can be consumed 1 time
-    /// or can be cloned and re-consumed
-    pub fn new<T>(msg: T, one_time: bool) -> Self
+    pub fn new<T>(msg: T) -> Self
     where
         T: Any + Message,
     {
         Self {
-            one_time,
             msg: Some(Box::new(msg)),
         }
     }
 
     /// Try and take the resulting message as a specific type, consumes
-    /// the boxed message if one_time = true
-    pub fn take<T>(&mut self) -> Result<T, DowncastBoxedMessageErr>
+    /// the boxed message
+    pub fn take<T>(&mut self) -> Result<T, BoxedDowncastErr>
     where
         T: Any + Message,
     {
-        if self.one_time {
-            match self.msg.take() {
-                Some(m) => {
-                    if m.is::<T>() {
-                        Ok(*m.downcast::<T>().unwrap())
-                    } else {
-                        Err(DowncastBoxedMessageErr)
-                    }
+        match self.msg.take() {
+            Some(m) => {
+                if m.is::<T>() {
+                    Ok(*m.downcast::<T>().unwrap())
+                } else {
+                    Err(BoxedDowncastErr)
                 }
-                None => Err(DowncastBoxedMessageErr),
-            }
-        } else {
-            match self.msg.as_ref() {
-                Some(m) if m.is::<T>() => Ok(m.downcast_ref::<T>().cloned().unwrap()),
-                Some(_) => Err(DowncastBoxedMessageErr),
-                None => Err(DowncastBoxedMessageErr),
             }
+            None => Err(BoxedDowncastErr),
         }
     }
 }
 
-impl Clone for BoxedMessage {
-    fn clone(&self) -> Self {
-        panic!("Cloning of an `BoxedMessage` type is not supported");
+/// A "boxed" state denoting a strong-type state
+/// but generic so it can be passed around without type
+/// constraints
+#[derive(Debug)]
+pub struct BoxedState {
+    /// The state value
+    pub state: Box<dyn Any + Send>,
+}
+
+impl BoxedState {
+    /// Create a new [BoxedState] from a strongly-typed state
+    pub fn new<T>(msg: T) -> Self
+    where
+        T: Any + State,
+    {
+        Self {
+            state: Box::new(msg),
+        }
+    }
+
+    /// Try and take the resulting type via cloning, such that we don't
+    /// consume the value
+    pub fn take<T>(&self) -> Result<T, BoxedDowncastErr>
+    where
+        T: Any + State,
+    {
+        let state_ref = &self.state;
+        if state_ref.is::<T>() {
+            Ok(state_ref.downcast_ref::<T>().cloned().unwrap())
+        } else {
+            Err(BoxedDowncastErr)
+        }
     }
 }
 
@@ -77,16 +94,44 @@ impl Debug for BoxedMessage {
 }
 
 /// A supervision event from the supervision tree
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub enum SupervisionEvent {
     /// An actor was started
     ActorStarted(super::actor_cell::ActorCell),
-    /// An actor terminated
-    ActorTerminated(super::actor_cell::ActorCell),
+    /// An actor terminated. In the event it shutdown cleanly (i.e. didn't panic or get
+    /// signaled) we capture the last state of the actor which can be used to re-build an actor
+    /// should the need arise
+    ActorTerminated(super::actor_cell::ActorCell, Option<BoxedState>),
     /// An actor panicked
     ActorPanicked(super::actor_cell::ActorCell, String),
 }
 
+impl SupervisionEvent {
+    /// A custom "clone" for [SupervisionEvent]s, because they hold a handle to a [BoxedState]
+    /// which can't directly implement the clone trait, therefore the [SupervisionEvent]
+    /// can't implement [Clone]. This requires that you give the intended strongly typed [State]
+    /// which will downcast, clone the underlying type, and create a new boxed state for you
+    pub(crate) fn duplicate<TState>(&self) -> Result<Self, BoxedDowncastErr>
+    where
+        TState: State,
+    {
+        match self {
+            Self::ActorStarted(actor) => Ok(Self::ActorStarted(actor.clone())),
+            Self::ActorTerminated(actor, maybe_state) => {
+                let cloned_maybe_state = match maybe_state {
+                    Some(state) => Some(state.take::<TState>()?),
+                    _ => None,
+                }
+                .map(BoxedState::new);
+                Ok(Self::ActorTerminated(actor.clone(), cloned_maybe_state))
+            }
+            Self::ActorPanicked(actor, message) => {
+                Ok(Self::ActorPanicked(actor.clone(), message.clone()))
+            }
+        }
+    }
+}
+
 /// A signal message which takes priority above all else
 #[derive(Clone, Debug)]
 pub enum Signal {
diff --git a/ractor/src/actor/mod.rs b/ractor/src/actor/mod.rs
index 7264b75..c6afc9d 100644
--- a/ractor/src/actor/mod.rs
+++ b/ractor/src/actor/mod.rs
@@ -21,7 +21,7 @@ mod tests;
 
 use crate::{Message, State};
 use actor_cell::{ActorCell, ActorPortSet, ActorStatus};
-use errors::{ActorProcessingErr, SpawnErr};
+use errors::{ActorErr, MessagingErr, SpawnErr};
 
 pub(crate) fn get_panic_string(e: Box<dyn std::any::Any + Send>) -> String {
     match e.downcast::<String>() {
@@ -58,10 +58,11 @@ async fn handle_panickable<TResult>(
 
 /// The message handling implementation for an actor with a specific type of input message and state
 #[async_trait::async_trait]
-pub trait ActorHandler: Sync + Send + 'static {
-    /// The message type for this handler
+pub trait ActorHandler: Sized + Sync + Send + 'static {
+    /// The message type for this actor
     type Msg: Message;
-    /// The type of state this actor deals with
+
+    /// The type of state this actor manages internally
     type State: State;
 
     /// Invoked when an actor is being started by the system.
@@ -88,10 +89,14 @@ pub trait ActorHandler: Sync + Send + 'static {
     }
 
     /// Invoked after an actor has been stopped.
-    async fn post_stop(&self, _this_actor: ActorCell, _state: &Self::State) {}
+    ///
+    /// Returns: The last state (after doing any teardown) which might be necessary to re-create the actor
+    async fn post_stop(&self, _this_actor: ActorCell, _state: Self::State) -> Self::State {
+        _state
+    }
 
     /// Handle the incoming message in a basic event processing loop. Unhandled panic's will be captured and
-    /// treated as agent death in the supervision tree
+    /// treated as actor death in the supervision tree
     async fn handle(
         &self,
         _this_actor: ActorCell,
@@ -101,7 +106,7 @@ pub trait ActorHandler: Sync + Send + 'static {
         None
     }
 
-    /// Handle the incoming supervision event. Unhandled panic's will captured and treated as agent death in
+    /// Handle the incoming supervision event. Unhandled panic's will captured and treated as actor death in
     /// the supervision tree
     async fn handle_supervisor_evt(
         &self,
@@ -111,6 +116,11 @@ pub trait ActorHandler: Sync + Send + 'static {
     ) -> Option<Self::State> {
         None
     }
+
+    /// Send a message to the specified actor, which is strong-typed to the message handler's type
+    fn send_message(&self, this_actor: ActorCell, msg: Self::Msg) -> Result<(), MessagingErr> {
+        this_actor.send_message::<Self, Self::Msg>(msg)
+    }
 }
 
 /// The basic actor with a
@@ -120,7 +130,7 @@ pub trait ActorHandler: Sync + Send + 'static {
 pub struct Actor<TMsg, TState, THandler>
 where
     TMsg: Message,
-    TState: State,
+    TState: State + Clone,
     THandler: ActorHandler<Msg = TMsg, State = TState>,
 {
     base: ActorCell,
@@ -130,10 +140,10 @@ where
 impl<TMsg, TState, THandler> Actor<TMsg, TState, THandler>
 where
     TMsg: Message,
-    TState: State,
+    TState: State + Clone,
     THandler: ActorHandler<Msg = TMsg, State = TState>,
 {
-    /// Create a new 10-port agent with some handler implementation and initial state
+    /// Create a new actor with some handler implementation and initial state
     pub fn new(name: Option<String>, handler: THandler) -> (Self, ActorPortSet) {
         let (actor_cell, ports) = ActorCell::new(name);
         (
@@ -186,7 +196,7 @@ where
 
         // setup supervision
         if let Some(sup) = &supervisor {
-            sup.link(self.base.clone()).await;
+            sup.link(self.base.clone());
         }
 
         // run the processing loop, capturing panic's
@@ -197,27 +207,30 @@ where
                 match Self::processing_loop(ports, state, self.handler.clone(), self.base.clone())
                     .await
                 {
-                    Ok(_) => SupervisionEvent::ActorTerminated(myself.clone()),
+                    Ok(last_state) => SupervisionEvent::ActorTerminated(
+                        myself.clone(),
+                        Some(BoxedState::new(last_state)),
+                    ),
                     Err(actor_err) => match actor_err {
-                        ActorProcessingErr::Cancelled => {
-                            SupervisionEvent::ActorTerminated(myself.clone())
+                        ActorErr::Cancelled => {
+                            SupervisionEvent::ActorTerminated(myself.clone(), None)
                         }
-                        ActorProcessingErr::Panic(msg) => {
+                        ActorErr::Panic(msg) => {
                             SupervisionEvent::ActorPanicked(myself.clone(), msg)
                         }
                     },
                 };
 
             // terminate children
-            myself.terminate().await;
+            myself.terminate();
 
             // notify supervisors of the actor's death
-            let _ = myself.notify_supervisors(evt).await;
+            myself.notify_supervisors::<THandler, TState>(evt);
 
             myself.set_status(ActorStatus::Stopped);
             // signal received or process exited cleanly, we should already have "handled" the signal, so we can just terminate
             if let Some(sup) = supervisor {
-                sup.unlink(myself.clone()).await;
+                sup.unlink(myself.clone());
             }
         });
 
@@ -229,14 +242,13 @@ where
         state: TState,
         handler: Arc<THandler>,
         myself: ActorCell,
-    ) -> Result<(), ActorProcessingErr> {
+    ) -> Result<TState, ActorErr> {
         // perform the post-start, with supervision enabled
         let mut state = Self::do_post_start(myself.clone(), handler.clone(), state).await?;
 
         myself.set_status(ActorStatus::Running);
-        let _ = myself
-            .notify_supervisors(SupervisionEvent::ActorStarted(myself.clone()))
-            .await;
+        myself
+            .notify_supervisors::<THandler, TState>(SupervisionEvent::ActorStarted(myself.clone()));
 
         // let mut last_state = state.clone();
         let myself_clone = myself.clone();
@@ -262,8 +274,8 @@ where
 
         match handle_panickable(handle).await {
             PanickableResult::Ok(_) => (),
-            PanickableResult::Cancelled => return Err(ActorProcessingErr::Cancelled),
-            PanickableResult::Panic(msg) => return Err(ActorProcessingErr::Panic(msg)),
+            PanickableResult::Cancelled => return Err(ActorErr::Cancelled),
+            PanickableResult::Panic(msg) => return Err(ActorErr::Panic(msg)),
         }
 
         myself.set_status(ActorStatus::Stopping);
@@ -305,7 +317,10 @@ where
             let typed_msg = match msg.take() {
                 Ok(m) => m,
                 Err(_) => {
-                    panic!("Failed to convert message from `BoxedMessage` to `TMsg`")
+                    panic!(
+                        "Failed to convert message from `BoxedMessage` to `{}`",
+                        std::any::type_name::<TMsg>()
+                    )
                 }
             };
             // NOTE: We listen for the signal port again during the processing of async work in order
@@ -330,7 +345,7 @@ where
     async fn handle_signal(myself: ActorCell, signal: Signal) -> Signal {
         match &signal {
             Signal::Exit => {
-                myself.terminate().await;
+                myself.terminate();
             }
         }
         // signal's always bubble up
@@ -343,7 +358,6 @@ where
         handler: Arc<THandler>,
         message: Option<SupervisionEvent>,
     ) -> Option<TState> {
-        // TODO: process the specific supervision logic (pass to handler?)
         if let Some(evt) = message {
             let maybe_new_state = handler.handle_supervisor_evt(myself, evt, state).await;
             maybe_new_state
@@ -353,9 +367,8 @@ where
     }
 
     async fn do_pre_start(myself: ActorCell, handler: Arc<THandler>) -> Result<TState, SpawnErr> {
-        let handle = tokio::spawn(async move { handler.pre_start(myself).await });
-        let start_result = handle_panickable(handle).await;
-        // let start_result = panic::catch_unwind(AssertUnwindSafe(|| handler.pre_start(myself)));
+        let start_result =
+            handle_panickable(tokio::spawn(async move { handler.pre_start(myself).await })).await;
         match start_result {
             PanickableResult::Ok(state) => {
                 // intitialize the state
@@ -373,17 +386,19 @@ where
         myself: ActorCell,
         handler: Arc<THandler>,
         state: TState,
-    ) -> Result<TState, ActorProcessingErr> {
-        // TODO: ugh i hate cloning here :/ but the async move moves ownership of "state" into
+    ) -> Result<TState, ActorErr> {
+        // TODO: Not a fan of cloning here :/ but the async move moves ownership of "state" into
         // the async block
         let original_state = state.clone();
-        let handle = tokio::spawn(async move { handler.post_start(myself, &state).await });
-        let post_start_result = handle_panickable(handle).await;
+        let post_start_result = handle_panickable(tokio::spawn(async move {
+            handler.post_start(myself, &state).await
+        }))
+        .await;
         match post_start_result {
             PanickableResult::Ok(Some(new_state)) => Ok(new_state),
             PanickableResult::Ok(None) => Ok(original_state),
-            PanickableResult::Cancelled => Err(ActorProcessingErr::Cancelled),
-            PanickableResult::Panic(panic_information) => Err(ActorProcessingErr::Panic(format!(
+            PanickableResult::Cancelled => Err(ActorErr::Cancelled),
+            PanickableResult::Panic(panic_information) => Err(ActorErr::Panic(format!(
                 "Actor panicked in post_start with '{}'",
                 panic_information
             ))),
@@ -394,15 +409,15 @@ where
         myself: ActorCell,
         handler: Arc<THandler>,
         state: TState,
-    ) -> Result<(), ActorProcessingErr> {
+    ) -> Result<TState, ActorErr> {
         let post_stop_result = handle_panickable(tokio::spawn(async move {
-            handler.post_stop(myself, &state).await
+            handler.post_stop(myself, state).await
         }))
         .await;
         match post_stop_result {
-            PanickableResult::Ok(_) => Ok(()),
-            PanickableResult::Cancelled => Err(ActorProcessingErr::Cancelled),
-            PanickableResult::Panic(panic_information) => Err(ActorProcessingErr::Panic(format!(
+            PanickableResult::Ok(last_state) => Ok(last_state),
+            PanickableResult::Cancelled => Err(ActorErr::Cancelled),
+            PanickableResult::Panic(panic_information) => Err(ActorErr::Panic(format!(
                 "Actor panicked in post_start with '{}'",
                 panic_information
             ))),
diff --git a/ractor/src/actor/supervision.rs b/ractor/src/actor/supervision.rs
index bd94724..a03b501 100644
--- a/ractor/src/actor/supervision.rs
+++ b/ractor/src/actor/supervision.rs
@@ -5,89 +5,89 @@
 
 //! Supervision management logic
 
-use std::{collections::HashMap, sync::Arc};
+use std::sync::Arc;
 
-use tokio::sync::RwLock;
+use dashmap::DashMap;
 
 use super::{actor_cell::ActorCell, messages::SupervisionEvent};
-use crate::ActorId;
+use crate::{ActorHandler, ActorId};
 
 /// A supervision tree
 #[derive(Clone, Default)]
 pub struct SupervisionTree {
-    children: Arc<RwLock<HashMap<ActorId, ActorCell>>>,
-    parents: Arc<RwLock<HashMap<ActorId, ActorCell>>>,
+    children: Arc<DashMap<ActorId, ActorCell>>,
+    parents: Arc<DashMap<ActorId, ActorCell>>,
 }
 
 impl SupervisionTree {
     /// Push a child into the tere
-    pub async fn insert_child(&self, child: ActorCell) {
-        let mut guard = self.children.write().await;
-        guard.insert(child.get_id(), child.clone());
+    pub fn insert_child(&self, child: ActorCell) {
+        self.children.insert(child.get_id(), child);
     }
 
     /// Remove a specific actor from the supervision tree (e.g. actor died)
-    pub async fn remove_child(&self, child: ActorCell) {
-        let mut guard = self.children.write().await;
+    pub fn remove_child(&self, child: ActorCell) {
         let id = child.get_id();
-        if guard.contains_key(&id) {
-            guard.remove(&id);
+        match self.children.entry(id) {
+            dashmap::mapref::entry::Entry::Occupied(item) => {
+                item.remove();
+            }
+            dashmap::mapref::entry::Entry::Vacant(_) => {}
         }
     }
 
     /// Push a parent into the tere
-    pub async fn insert_parent(&self, parent: ActorCell) {
-        let mut guard = self.parents.write().await;
-        guard.insert(parent.get_id(), parent.clone());
+    pub fn insert_parent(&self, parent: ActorCell) {
+        self.parents.insert(parent.get_id(), parent);
     }
 
     /// Remove a specific actor from the supervision tree (e.g. actor died)
-    pub async fn remove_parent(&self, parent: ActorCell) {
-        let mut guard = self.parents.write().await;
+    pub fn remove_parent(&self, parent: ActorCell) {
         let id = parent.get_id();
-        if guard.contains_key(&id) {
-            guard.remove(&id);
+        match self.parents.entry(id) {
+            dashmap::mapref::entry::Entry::Occupied(item) => {
+                item.remove();
+            }
+            dashmap::mapref::entry::Entry::Vacant(_) => {}
         }
     }
 
     /// Terminate all your supervised children
-    #[async_recursion::async_recursion]
-    pub async fn terminate_children(&self) {
-        let guard = self.children.read().await;
-
-        for (_, child) in guard.iter() {
-            child.terminate().await;
+    pub fn terminate_children(&self) {
+        for kvp in self.children.iter() {
+            kvp.value().terminate();
         }
     }
 
     /// Determine if the specified actor is a member of this supervision tree
-    pub async fn is_supervisor_of(&self, id: ActorId) -> bool {
-        self.children.read().await.contains_key(&id)
+    pub fn is_supervisor_of(&self, id: ActorId) -> bool {
+        self.children.contains_key(&id)
     }
 
     /// Determine if the specified actor is a parent of this actor
-    pub async fn is_child_of(&self, id: ActorId) -> bool {
-        self.parents.read().await.contains_key(&id)
+    pub fn is_child_of(&self, id: ActorId) -> bool {
+        self.parents.contains_key(&id)
     }
 
     /// Send a notification to all supervisors
-    pub async fn notify_supervisors(
-        &self,
-        evt: SupervisionEvent,
-    ) -> Result<(), tokio::sync::mpsc::error::SendError<SupervisionEvent>> {
-        for (_, parent) in self.parents.read().await.iter() {
-            parent.send_supervisor_evt(evt.clone())?;
+    pub fn notify_supervisors<TActor, TState>(&self, evt: SupervisionEvent)
+    where
+        TActor: ActorHandler<State = TState>,
+        TState: crate::State,
+    {
+        for kvp in self.parents.iter() {
+            let evt_clone = evt.duplicate::<TState>().unwrap();
+            let _ = kvp.value().send_supervisor_evt(evt_clone);
         }
-        Ok(())
     }
 
     /// Retrieve the number of supervised children
-    pub async fn get_num_children(&self) -> usize {
-        self.children.read().await.len()
+    pub fn get_num_children(&self) -> usize {
+        self.children.len()
     }
 
     /// Retrieve the number of supervised children
-    pub async fn get_num_parents(&self) -> usize {
-        self.parents.read().await.len()
+    pub fn get_num_parents(&self) -> usize {
+        self.parents.len()
     }
 }
diff --git a/ractor/src/lib.rs b/ractor/src/lib.rs
index 44847f7..a8972cb 100644
--- a/ractor/src/lib.rs
+++ b/ractor/src/lib.rs
@@ -9,27 +9,73 @@
 #![warn(unused_crate_dependencies)]
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
-use core::fmt::Debug;
 use std::sync::atomic::AtomicU64;
 
-/// An agent's identifier
+/// An actor's globally unique identifier
 pub type ActorId = u64;
-/// The global id allocator for agents
-pub static AGENT_ID_ALLOCATOR: AtomicU64 = AtomicU64::new(0u64);
+/// The global id allocator for actors
+pub static ACTOR_ID_ALLOCATOR: AtomicU64 = AtomicU64::new(0u64);
 
 pub mod actor;
 pub mod port;
+pub mod rpc;
 
 // re-exports
 pub use actor::actor_cell::{ActorCell, ActorStatus};
-pub use actor::errors::{ActorProcessingErr, SpawnErr};
+pub use actor::errors::{ActorErr, MessagingErr, SpawnErr};
 pub use actor::messages::{Signal, SupervisionEvent};
 pub use actor::{Actor, ActorHandler};
+pub use port::{InputPort, RpcReplyPort};
 
-/// Message type for an agent
-pub trait Message: Debug + Clone + Send + 'static {}
-impl<T: Debug + Clone + Send + 'static> Message for T {}
+/// Message type for an actor
+pub trait Message: Send + 'static {}
+impl<T: Send + 'static> Message for T {}
 
-/// An agent's state
-pub trait State: Clone + Sync + Send + std::any::Any {}
-impl<T: Clone + Sync + Send + std::any::Any> State for T {}
+/// Represents the state of an actor
+pub trait State: Clone + Sync + Send + 'static {}
+impl<T: Clone + Sync + Send + 'static> State for T {}
+
+/// Error types which can result from Ractor processes
+#[derive(Debug)]
+pub enum RactorErr {
+    /// An error occurred spawning
+    Spawn(SpawnErr),
+    /// An error occurred in messaging (sending/receiving)
+    Messaging(MessagingErr),
+    /// An actor encountered an error while processing (canceled or panicked)
+    Actor(ActorErr),
+}
+
+impl From<SpawnErr> for RactorErr {
+    fn from(value: SpawnErr) -> Self {
+        RactorErr::Spawn(value)
+    }
+}
+
+impl From<MessagingErr> for RactorErr {
+    fn from(value: MessagingErr) -> Self {
+        RactorErr::Messaging(value)
+    }
+}
+
+impl From<ActorErr> for RactorErr {
+    fn from(value: ActorErr) -> Self {
+        RactorErr::Actor(value)
+    }
+}
+
+impl std::fmt::Display for RactorErr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Actor(actor_err) => {
+                write!(f, "{}", actor_err)
+            }
+            Self::Messaging(messaging_err) => {
+                write!(f, "{}", messaging_err)
+            }
+            Self::Spawn(spawn_err) => {
+                write!(f, "{}", spawn_err)
+            }
+        }
+    }
+}
diff --git a/ractor/src/port/input.rs b/ractor/src/port/input.rs
deleted file mode 100644
index 812d942..0000000
--- a/ractor/src/port/input.rs
+++ /dev/null
@@ -1,11 +0,0 @@
-// Copyright (c) Sean Lawlor
-//
-// This source code is licensed under both the MIT license found in the
-// LICENSE-MIT file in the root directory of this source tree.
-
-//! Input port implementations for various sizes
-
-/// An import-port
-pub type InputPort<TMsg> = tokio::sync::mpsc::UnboundedSender<TMsg>;
-
-pub(crate) type InputPortReceiver<TMsg> = tokio::sync::mpsc::UnboundedReceiver<TMsg>;
diff --git a/ractor/src/port/mod.rs b/ractor/src/port/mod.rs
index eb845c8..777609a 100644
--- a/ractor/src/port/mod.rs
+++ b/ractor/src/port/mod.rs
@@ -5,29 +5,50 @@
 
 //! Port implementations for signaling and reception of messages in the Ractor environment
 
-pub mod input;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+
+use crate::MessagingErr;
+
+// ============ Input Ports ============ //
+
+/// A bounded imput port (alias of [mpsc::Sender])
+pub(crate) type BoundedInputPort<TMsg> = mpsc::Sender<TMsg>;
+/// A bounded imput port (alias of [mpsc::Receiver])
+pub(crate) type BoundedInputPortReceiver<TMsg> = mpsc::Receiver<TMsg>;
+
+/// An import-port (alias of [mpsc::UnboundedSender])
+pub type InputPort<TMsg> = mpsc::UnboundedSender<TMsg>;
+/// An input port's receiver (alias of [mpsc::UnboundedReceiver])
+pub(crate) type InputPortReceiver<TMsg> = mpsc::UnboundedReceiver<TMsg>;
+
+// ============ Output Ports ============ //
+// TODO: outputs with auto-subscriptions
 // pub mod output;
+// pub use output::*;
+
+// ============ Rpc (one-use) Ports ============ //
+
+/// A RPC's reply port. Wrapper of [tokio::sync::oneshot::Sender] with a consistent
+/// error type
+pub struct RpcReplyPort<TMsg> {
+    port: oneshot::Sender<TMsg>,
+}
+
+impl<TMsg> RpcReplyPort<TMsg> {
+    /// Send a message to the Rpc reply port. This consumes the port
+    pub fn send(self, msg: TMsg) -> Result<(), MessagingErr> {
+        self.port.send(msg).map_err(|_| MessagingErr::ChannelClosed)
+    }
+
+    /// Determine if the port is closed (i.e. the receiver has been dropped)
+    pub fn is_closed(&self) -> bool {
+        self.port.is_closed()
+    }
+}
 
-// /// A muxed value over the wire sent through a [MuxedPort]
-// pub enum MuxedValue<T1, T2, T3, T4, T5, T6, T7, T8, T9, T10> {
-//     /// Type 1
-//     T1(T1),
-//     /// Type 2
-//     T2(T2),
-//     /// Type 3
-//     T3(T3),
-//     /// Type 4
-//     T4(T4),
-//     /// Type 5
-//     T5(T5),
-//     /// Type 6
-//     T6(T6),
-//     /// Type 7
-//     T7(T7),
-//     /// Type 8
-//     T8(T8),
-//     /// Type 9
-//     T9(T9),
-//     /// Type 10
-//     T10(T10),
-// }
+impl<TMsg> From<oneshot::Sender<TMsg>> for RpcReplyPort<TMsg> {
+    fn from(value: oneshot::Sender<TMsg>) -> Self {
+        Self { port: value }
+    }
+}
diff --git a/ractor/src/rpc/call_result.rs b/ractor/src/rpc/call_result.rs
new file mode 100644
index 0000000..d5a9347
--- /dev/null
+++ b/ractor/src/rpc/call_result.rs
@@ -0,0 +1,135 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! This module contains the remote procedure call [CallResult] structure and supported operations
+
+/// The result from a [crate::rpc::call] operation
+#[derive(Debug, Eq, PartialEq)]
+pub enum CallResult<TResult> {
+    /// Success, with the result
+    Success(TResult),
+    /// Timeout
+    Timeout,
+    /// The transmission channel was dropped without any message(s) being sent
+    SenderError,
+}
+
+impl<T> CallResult<T> {
+    /// Determine if the [CallResult] is a [CallResult::Success]
+    pub fn is_success(&self) -> bool {
+        matches!(self, Self::Success(_))
+    }
+
+    /// Determin if the [CallResult] is a [CallResult::Timeout]
+    pub fn is_timeout(&self) -> bool {
+        matches!(self, Self::Timeout)
+    }
+
+    /// Determin if the [CallResult] is a [CallResult::SenderError]
+    pub fn is_send_error(&self) -> bool {
+        matches!(self, Self::SenderError)
+    }
+
+    /// Unwrap a [CallResult], panicking on any non-success
+    pub fn unwrap(self) -> T {
+        match self {
+            Self::Success(t) => t,
+            Self::Timeout => panic!("called CallResult::<T>::unwrap()  on a `Timeout` value"),
+            Self::SenderError => {
+                panic!("called CallResult::<T>::unwrap() on a `SenderError` value")
+            }
+        }
+    }
+
+    /// Unwrap a [CallResult], panicking on non-succcess with the specified message
+    pub fn expect(self, msg: &'static str) -> T {
+        match self {
+            Self::Success(t) => t,
+            Self::Timeout => panic!("{}", msg),
+            Self::SenderError => panic!("{}", msg),
+        }
+    }
+
+    /// Unwrap the [CallResult] or give a default value
+    pub fn unwrap_or(self, default: T) -> T {
+        if let Self::Success(t) = self {
+            t
+        } else {
+            default
+        }
+    }
+
+    /// Returns the [CallResult]'s success result or computes the closure
+    pub fn unwrap_or_else<F>(self, f: F) -> T
+    where
+        F: FnOnce() -> T,
+    {
+        if let Self::Success(t) = self {
+            t
+        } else {
+            f()
+        }
+    }
+
+    /// Tranforms the [CallResult] to a Result mapping `Success(t)` to `Ok(t)` and all else to `Err(err)`
+    pub fn success_or<E>(self, err: E) -> Result<T, E> {
+        if let Self::Success(t) = self {
+            Ok(t)
+        } else {
+            Err(err)
+        }
+    }
+
+    /// Tranforms the [CallResult] to a Result mapping `Success(t)` to `Ok(t)` and all else to `Err(err())`
+    pub fn success_or_else<E, F>(self, err: F) -> Result<T, E>
+    where
+        F: FnOnce() -> E,
+    {
+        if let Self::Success(t) = self {
+            Ok(t)
+        } else {
+            Err(err())
+        }
+    }
+
+    /// Maps the success value of the [CallResult] to another type
+    pub fn map<O, F>(self, mapping: F) -> CallResult<O>
+    where
+        F: FnOnce(T) -> O,
+    {
+        match self {
+            Self::Success(t) => CallResult::Success(mapping(t)),
+            Self::Timeout => CallResult::Timeout,
+            Self::SenderError => CallResult::SenderError,
+        }
+    }
+
+    /// Maps the success value of the [CallResult] to another type
+    /// or returns the default value
+    pub fn map_or<O, F>(self, default: O, mapping: F) -> O
+    where
+        F: FnOnce(T) -> O,
+    {
+        match self {
+            Self::Success(t) => mapping(t),
+            Self::Timeout => default,
+            Self::SenderError => default,
+        }
+    }
+
+    /// Maps the success value of the [CallResult] to another type
+    /// or returns the default function result
+    pub fn map_or_else<D, O, F>(self, default: D, mapping: F) -> O
+    where
+        F: FnOnce(T) -> O,
+        D: FnOnce() -> O,
+    {
+        match self {
+            Self::Success(t) => mapping(t),
+            Self::Timeout => default(),
+            Self::SenderError => default(),
+        }
+    }
+}
diff --git a/ractor/src/rpc/mod.rs b/ractor/src/rpc/mod.rs
new file mode 100644
index 0000000..78c124f
--- /dev/null
+++ b/ractor/src/rpc/mod.rs
@@ -0,0 +1,108 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! Helpers for remote-procedure calls
+
+use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
+use tokio::time::{self, Duration};
+
+use crate::{ActorCell, ActorHandler, Message, MessagingErr, RpcReplyPort};
+
+pub mod call_result;
+pub use call_result::CallResult;
+#[cfg(test)]
+mod tests;
+
+/// Sends an asynchronous request to the specified actor, ignoring if the
+/// actor is alive or healthy and simply returns immediately
+pub fn cast<TActor, TMsg>(actor: &ActorCell, msg: TMsg) -> Result<(), MessagingErr>
+where
+    TActor: ActorHandler<Msg = TMsg>,
+    TMsg: Message,
+{
+    actor.send_message::<TActor, _>(msg)
+}
+
+/// Sends an asynchronous request to the specified actor, building a one-time
+/// use reply channel and awaiting the result with the specified timeout
+pub async fn call<TActor, TMsg, TReply, TMsgBuilder>(
+    actor: &ActorCell,
+    msg_builder: TMsgBuilder,
+    timeout_option: Option<Duration>,
+) -> Result<CallResult<TReply>, MessagingErr>
+where
+    TActor: ActorHandler<Msg = TMsg>,
+    TMsg: Message,
+    TMsgBuilder: FnOnce(RpcReplyPort<TReply>) -> TMsg,
+{
+    let (tx, rx) = oneshot::channel();
+    actor.send_message::<TActor, _>(msg_builder(tx.into()))?;
+
+    // wait for the reply
+    Ok(if let Some(duration) = timeout_option {
+        match time::timeout(duration, rx).await {
+            Ok(Ok(result)) => CallResult::Success(result),
+            Ok(Err(_send_err)) => CallResult::SenderError,
+            Err(_timeout_err) => CallResult::Timeout,
+        }
+    } else {
+        match rx.await {
+            Ok(result) => CallResult::Success(result),
+            Err(_send_err) => CallResult::SenderError,
+        }
+    })
+}
+
+/// Send a message asynchronously to another actor, waiting in a new task for the reply
+/// and then forwaring the reply to a followup-actor. If this [CallResult] from the first
+/// actor is not success, the forward is not sent.
+///
+/// Returns: A [JoinHandle<CallResult<()>>] which can be awaited to see if the
+/// forward was successful or ignored
+pub fn call_and_forward<
+    TActor,
+    TForwardActor,
+    TMsg,
+    TReply,
+    TMsgBuilder,
+    FwdMapFn,
+    TForwardMessage,
+>(
+    actor: &ActorCell,
+    msg_builder: TMsgBuilder,
+    response_forward: ActorCell,
+    forward_mapping: FwdMapFn,
+    timeout_option: Option<Duration>,
+) -> Result<JoinHandle<CallResult<Result<(), MessagingErr>>>, MessagingErr>
+where
+    TActor: ActorHandler<Msg = TMsg>,
+    TMsg: Message,
+    TReply: Message,
+    TMsgBuilder: FnOnce(RpcReplyPort<TReply>) -> TMsg,
+    TForwardActor: ActorHandler<Msg = TForwardMessage>,
+    FwdMapFn: FnOnce(TReply) -> TForwardMessage + Send + 'static,
+    TForwardMessage: Message,
+{
+    let (tx, rx) = oneshot::channel();
+    actor.send_message::<TActor, _>(msg_builder(tx.into()))?;
+
+    // wait for the reply
+    Ok(tokio::spawn(async move {
+        if let Some(duration) = timeout_option {
+            match time::timeout(duration, rx).await {
+                Ok(Ok(result)) => CallResult::Success(result),
+                Ok(Err(_send_err)) => CallResult::SenderError,
+                Err(_timeout_err) => CallResult::Timeout,
+            }
+        } else {
+            match rx.await {
+                Ok(result) => CallResult::Success(result),
+                Err(_send_err) => CallResult::SenderError,
+            }
+        }
+        .map(|msg| response_forward.send_message::<TForwardActor, _>(forward_mapping(msg)))
+    }))
+}


Test patch:
diff --git a/ractor/src/actor/tests/supervisor.rs b/ractor/src/actor/tests/supervisor.rs
index 67101fe..28629de 100644
--- a/ractor/src/actor/tests/supervisor.rs
+++ b/ractor/src/actor/tests/supervisor.rs
@@ -10,7 +10,9 @@ use std::sync::{
     Arc,
 };
 
-use crate::{Actor, ActorCell, ActorHandler, SupervisionEvent};
+use tokio::time::Duration;
+
+use crate::{Actor, ActorCell, ActorHandler, ActorStatus, SupervisionEvent};
 
 #[tokio::test]
 async fn test_supervision_panic_in_post_startup() {
@@ -23,9 +25,7 @@ async fn test_supervision_panic_in_post_startup() {
     impl ActorHandler for Child {
         type Msg = ();
         type State = ();
-        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {
-            ()
-        }
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {}
         async fn post_start(
             &self,
             _this_actor: ActorCell,
@@ -60,7 +60,7 @@ async fn test_supervision_panic_in_post_startup() {
             // check that the panic was captured
             if let SupervisionEvent::ActorPanicked(dead_actor, _panic_msg) = message {
                 self.flag.store(dead_actor.get_id(), Ordering::Relaxed);
-                let _ = this_actor.stop().await;
+                this_actor.stop();
             }
 
             None
@@ -88,7 +88,7 @@ async fn test_supervision_panic_in_post_startup() {
     assert_eq!(child_ref.get_id(), flag.load(Ordering::Relaxed));
 
     // supervisor relationship cleaned up correctly
-    assert_eq!(0, supervisor_tree.get_num_children().await);
+    assert_eq!(0, supervisor_tree.get_num_children());
 }
 
 #[tokio::test]
@@ -138,7 +138,7 @@ async fn test_supervision_panic_in_handle() {
             // check that the panic was captured
             if let SupervisionEvent::ActorPanicked(dead_actor, _panic_msg) = message {
                 self.flag.store(dead_actor.get_id(), Ordering::Relaxed);
-                let _ = this_actor.stop().await;
+                this_actor.stop();
             }
 
             None
@@ -162,20 +162,20 @@ async fn test_supervision_panic_in_handle() {
         .expect("Child panicked on startup");
 
     // check that the supervision is wired up correctly
-    assert_eq!(1, supervisor_tree.get_num_children().await);
-    assert_eq!(1, child_ref.get_tree().get_num_parents().await);
+    assert_eq!(1, supervisor_tree.get_num_children());
+    assert_eq!(1, child_ref.get_tree().get_num_parents());
 
     // trigger the child failure
     child_ref
-        .send_message_t(())
-        .expect("Failed to send message to child");
+        .send_message::<Child, _>(())
+        .expect("Failed to send message");
 
     let (_, _) = tokio::join!(s_handle, c_handle);
 
     assert_eq!(child_ref.get_id(), flag.load(Ordering::Relaxed));
 
     // supervisor relationship cleaned up correctly
-    assert_eq!(0, supervisor_tree.get_num_children().await);
+    assert_eq!(0, supervisor_tree.get_num_children());
 }
 
 #[tokio::test]
@@ -191,10 +191,9 @@ async fn test_supervision_panic_in_post_stop() {
         type State = ();
         async fn pre_start(&self, this_actor: ActorCell) -> Self::State {
             // trigger stop, which starts shutdown
-            let _ = this_actor.stop().await;
-            ()
+            this_actor.stop();
         }
-        async fn post_stop(&self, _this_actor: ActorCell, _state: &Self::State) {
+        async fn post_stop(&self, _this_actor: ActorCell, _state: Self::State) -> Self::State {
             panic!("Boom");
         }
     }
@@ -224,7 +223,7 @@ async fn test_supervision_panic_in_post_stop() {
             // check that the panic was captured
             if let SupervisionEvent::ActorPanicked(dead_actor, _panic_msg) = message {
                 self.flag.store(dead_actor.get_id(), Ordering::Relaxed);
-                let _ = this_actor.stop().await;
+                this_actor.stop();
             }
 
             None
@@ -252,7 +251,7 @@ async fn test_supervision_panic_in_post_stop() {
     assert_eq!(child_ref.get_id(), flag.load(Ordering::Relaxed));
 
     // supervisor relationship cleaned up correctly
-    assert_eq!(0, supervisor_tree.get_num_children().await);
+    assert_eq!(0, supervisor_tree.get_num_children());
 }
 
 /// Test that a panic in the supervisor's handling propogates to
@@ -323,7 +322,7 @@ async fn test_supervision_panic_in_supervisor_handle() {
             // check that the panic was captured
             if let SupervisionEvent::ActorPanicked(dead_actor, _panic_msg) = message {
                 self.flag.store(dead_actor.get_id(), Ordering::Relaxed);
-                let _ = this_actor.stop().await;
+                this_actor.stop();
             }
 
             None
@@ -356,19 +355,19 @@ async fn test_supervision_panic_in_supervisor_handle() {
         .expect("Child panicked on startup");
 
     // check that the supervision is wired up correctly
-    assert_eq!(1, supervisor_tree.get_num_children().await);
-    assert_eq!(1, midpoint_tree.get_num_children().await);
-    assert_eq!(1, midpoint_tree.get_num_parents().await);
-    assert_eq!(1, child_ref.get_tree().get_num_parents().await);
+    assert_eq!(1, supervisor_tree.get_num_children());
+    assert_eq!(1, midpoint_tree.get_num_children());
+    assert_eq!(1, midpoint_tree.get_num_parents());
+    assert_eq!(1, child_ref.get_tree().get_num_parents());
 
     // trigger the child failure
     child_ref
-        .send_message_t(())
-        .expect("Failed to send message to child");
+        .send_message::<Child, _>(())
+        .expect("Failed to send message");
 
     // which triggers the handler in the midpoint, which panic's in supervision
 
-    // wait for all agents to die in the test
+    // wait for all actors to die in the test
     let _ = s_handle.await;
     let _ = c_handle.await;
     let _ = m_handle.await;
@@ -377,5 +376,69 @@ async fn test_supervision_panic_in_supervisor_handle() {
     assert_eq!(midpoint_ref_clone.get_id(), flag.load(Ordering::Relaxed));
 
     // supervisor relationship cleaned up correctly
-    assert_eq!(0, supervisor_tree.get_num_children().await);
+    assert_eq!(0, supervisor_tree.get_num_children());
+}
+
+#[tokio::test]
+async fn test_killing_a_supervisor_terminates_children() {
+    struct Child;
+    struct Supervisor;
+
+    #[async_trait::async_trait]
+    impl ActorHandler for Child {
+        type Msg = ();
+        type State = ();
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {}
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for Supervisor {
+        type Msg = ();
+        type State = ();
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {}
+        async fn handle(
+            &self,
+            _this_actor: ActorCell,
+            _message: Self::Msg,
+            _state: &Self::State,
+        ) -> Option<Self::State> {
+            // stop the supervisor, which starts the supervision shutdown of children
+            _this_actor.stop();
+            None
+        }
+    }
+
+    let (supervisor, supervisor_ports) = Actor::new(None, Supervisor);
+    let (supervisor_ref, s_handle) = supervisor
+        .start(supervisor_ports, None)
+        .await
+        .expect("Supervisor panicked on startup");
+
+    let supervisor_tree = supervisor_ref.get_tree();
+
+    let (child, child_ports) = Actor::new(None, Child);
+    let (child_ref, c_handle) = child
+        .start(child_ports, Some(supervisor_ref.clone()))
+        .await
+        .expect("Child panicked on startup");
+
+    // check that the supervision is wired up correctly
+    assert_eq!(1, supervisor_tree.get_num_children());
+    assert_eq!(1, child_ref.get_tree().get_num_parents());
+
+    // initate the shutdown of the supervisor
+    supervisor_ref
+        .cast::<Supervisor, _>(())
+        .expect("Sending message to supervisor failed");
+
+    s_handle
+        .await
+        .expect("Failed to wait for supervisor to shutdown");
+
+    // wait for async shutdown
+    tokio::time::sleep(Duration::from_millis(50)).await;
+
+    assert_eq!(ActorStatus::Stopped, child_ref.get_status());
+
+    c_handle.await.expect("Failed to wait for child to die");
 }
diff --git a/ractor/src/rpc/tests.rs b/ractor/src/rpc/tests.rs
new file mode 100644
index 0000000..59edbf6
--- /dev/null
+++ b/ractor/src/rpc/tests.rs
@@ -0,0 +1,256 @@
+// Copyright (c) Sean Lawlor
+//
+// This source code is licensed under both the MIT license found in the
+// LICENSE-MIT file in the root directory of this source tree.
+
+//! Tests for remote procedure calls
+
+use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::Arc;
+
+use tokio::time::Duration;
+
+use crate::rpc;
+use crate::{Actor, ActorCell, ActorHandler};
+
+#[tokio::test]
+async fn test_rpc_cast() {
+    let counter = Arc::new(AtomicU8::new(0u8));
+
+    struct TestActor {
+        counter: Arc<AtomicU8>,
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for TestActor {
+        type Msg = ();
+
+        type State = ();
+
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {}
+
+        async fn handle(
+            &self,
+            _this_actor: ActorCell,
+            _message: Self::Msg,
+            _state: &Self::State,
+        ) -> Option<Self::State> {
+            self.counter.fetch_add(1u8, Ordering::Relaxed);
+            None
+        }
+    }
+
+    let (actor, ports) = Actor::new(
+        None,
+        TestActor {
+            counter: counter.clone(),
+        },
+    );
+    let (actor_ref, handle) = actor
+        .start(ports, None)
+        .await
+        .expect("Failed to start test actor");
+
+    rpc::cast::<TestActor, _>(&actor_ref, ()).expect("Failed to send message");
+    actor_ref
+        .cast::<TestActor, _>(())
+        .expect("Failed to send message");
+
+    // make sure they have time to process
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    // assert the actor received 2 cast requests
+    assert_eq!(2, counter.load(Ordering::Relaxed));
+
+    // cleanup
+    actor_ref.stop();
+    handle.await.expect("Actor stopped with err");
+}
+
+#[tokio::test]
+async fn test_rpc_call() {
+    struct TestActor;
+
+    enum MessageFormat {
+        TestRpc(rpc::RpcReplyPort<String>),
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for TestActor {
+        type Msg = MessageFormat;
+
+        type State = ();
+
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {}
+
+        async fn handle(
+            &self,
+            _this_actor: ActorCell,
+            message: Self::Msg,
+            _state: &Self::State,
+        ) -> Option<Self::State> {
+            match message {
+                Self::Msg::TestRpc(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());
+                    }
+                }
+            }
+            None
+        }
+    }
+
+    let (actor, ports) = Actor::new(None, TestActor);
+    let (actor_ref, handle) = actor
+        .start(ports, None)
+        .await
+        .expect("Failed to start test actor");
+
+    let rpc_result = rpc::call::<TestActor, _, _, _>(
+        &actor_ref,
+        MessageFormat::TestRpc,
+        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_result = actor_ref
+        .call::<TestActor, _, _, _>(MessageFormat::TestRpc, 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);
+
+    // cleanup
+    actor_ref.stop();
+    handle.await.expect("Actor stopped with err");
+}
+
+#[tokio::test]
+async fn test_rpc_call_forwarding() {
+    struct Worker;
+
+    enum WorkerMessage {
+        TestRpc(rpc::RpcReplyPort<String>),
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for Worker {
+        type Msg = WorkerMessage;
+
+        type State = ();
+
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {}
+
+        async fn handle(
+            &self,
+            _this_actor: ActorCell,
+            message: Self::Msg,
+            _state: &Self::State,
+        ) -> Option<Self::State> {
+            match message {
+                Self::Msg::TestRpc(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());
+                    }
+                }
+            }
+            None
+        }
+    }
+
+    let counter = Arc::new(AtomicU8::new(0u8));
+    struct Forwarder {
+        counter: Arc<AtomicU8>,
+    }
+
+    enum ForwarderMessage {
+        ForwardResult(String),
+    }
+
+    #[async_trait::async_trait]
+    impl ActorHandler for Forwarder {
+        type Msg = ForwarderMessage;
+
+        type State = ();
+
+        async fn pre_start(&self, _this_actor: ActorCell) -> Self::State {}
+
+        async fn handle(
+            &self,
+            _this_actor: ActorCell,
+            message: Self::Msg,
+            _state: &Self::State,
+        ) -> Option<Self::State> {
+            match message {
+                Self::Msg::ForwardResult(s) if s == *"howdy" => {
+                    self.counter.fetch_add(1, Ordering::Relaxed);
+                }
+                _ => {}
+            }
+            None
+        }
+    }
+
+    let (worker, ports) = Actor::new(None, Worker);
+    let (worker_ref, worker_handle) = worker
+        .start(ports, None)
+        .await
+        .expect("Failed to start worker actor");
+
+    let (forwarder, ports) = Actor::new(
+        None,
+        Forwarder {
+            counter: counter.clone(),
+        },
+    );
+    let (forwarder_ref, forwarder_handle) = forwarder
+        .start(ports, None)
+        .await
+        .expect("Failed to start forwarder actor");
+
+    let forward_handle = rpc::call_and_forward::<Worker, Forwarder, _, _, _, _, _>(
+        &worker_ref,
+        WorkerMessage::TestRpc,
+        forwarder_ref.clone(),
+        ForwarderMessage::ForwardResult,
+        Some(Duration::from_millis(100)),
+    );
+
+    forward_handle
+        .expect("Failed to send message to actor")
+        .await
+        .expect("Forwarding task cancelled or panicked")
+        .expect("Call result didn't return success")
+        .expect("Failed to forward message");
+
+    let forward_handle = worker_ref.call_and_forward::<Worker, Forwarder, _, _, _, _, _>(
+        WorkerMessage::TestRpc,
+        forwarder_ref.clone(),
+        ForwarderMessage::ForwardResult,
+        Some(Duration::from_millis(100)),
+    );
+
+    forward_handle
+        .expect("Failed to send message to actor")
+        .await
+        .expect("Forwarding task cancelled or panicked")
+        .expect("Call result didn't return success")
+        .expect("Failed to forward message");
+
+    // make sure the counter was bumped to say the message was forwarded
+    assert_eq!(2, counter.load(Ordering::Relaxed));
+
+    // cleanup
+    forwarder_ref.stop();
+    worker_ref.stop();
+
+    forwarder_handle.await.expect("Actor stopped with err");
+    worker_handle.await.expect("Actor stopped with err");
+}