test callers and senders

examples/caller.rs

@@ -0,0 +1,43 @@
+use xactor::*;
+
+/// Define `Ping` message
+#[message(result = "usize")]
+struct Ping(usize);
+
+/// Actor
+struct MyActor {
+    count: usize,
+}
+
+/// Declare actor and its context
+impl Actor for MyActor {}
+
+/// Handler for `Ping` message
+#[async_trait::async_trait]
+impl Handler<Ping> for MyActor {
+    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: Ping) -> usize {
+        self.count += msg.0;
+        self.count
+    }
+}
+
+#[xactor::main]
+async fn main() -> Result<()> {
+    // start new actor
+    let addr = MyActor { count: 10 }.start().await?;
+
+    let caller: Caller<Ping> = addr.caller();
+
+    let res = caller.call(Ping(10)).await?;
+    println!("RESULT: {}", res == 20);
+
+
+    println!("caller can upgrade: {}", caller.can_upgrade());
+    std::mem::drop(addr);
+    println!("caller can upgrade: {}", caller.can_upgrade());
+
+    let res = caller.call(Ping(10)).await?;
+    println!("RESULT: {}", res == 20);
+
+    Ok(())
+}

src/addr.rs

@@ -1,3 +1,4 @@
+use crate::caller::CallerFn;
 use crate::{Actor, ActorId, Caller, Context, Error, Handler, Message, Result, Sender};
 use futures::channel::{mpsc, oneshot};
 use futures::future::Shared;
@@ -110,30 +111,35 @@ impl<A: Actor> Addr<A> {
         A: Handler<T>,
     {
         let weak_tx = Arc::downgrade(&self.tx);
+        let caller_fn: Mutex<CallerFn<T>> = Mutex::new(Box::new(move |msg| {
+            let weak_tx_option = weak_tx.upgrade();
+            Box::pin(async move {
+                match weak_tx_option {
+                    Some(tx) => {
+                        let (oneshot_tx, oneshot_rx) = oneshot::channel();
+
+                        mpsc::UnboundedSender::clone(&tx).start_send(ActorEvent::Exec(
+                            Box::new(move |actor, ctx| {
+                                Box::pin(async move {
+                                    let res = Handler::handle(&mut *actor, ctx, msg).await;
+                                    let _ = oneshot_tx.send(res);
+                                })
+                            }),
+                        ))?;
+                        Ok(oneshot_rx.await?)
+                    }
+                    None => Err(crate::error::anyhow!("Actor Dropped")),
+                }
+            })
+        }));
+
+        let weak_tx = Arc::downgrade(&self.tx);
+        let test_fn = Box::new(move || weak_tx.strong_count() > 0);
 
         Caller {
-            actor_id: self.actor_id.clone(),
-            caller_fn: Mutex::new(Box::new(move |msg| {
-                let weak_tx_option = weak_tx.upgrade();
-                Box::pin(async move {
-                    match weak_tx_option {
-                        Some(tx) => {
-                            let (oneshot_tx, oneshot_rx) = oneshot::channel();
-
-                            mpsc::UnboundedSender::clone(&tx).start_send(ActorEvent::Exec(
-                                Box::new(move |actor, ctx| {
-                                    Box::pin(async move {
-                                        let res = Handler::handle(&mut *actor, ctx, msg).await;
-                                        let _ = oneshot_tx.send(res);
-                                    })
-                                }),
-                            ))?;
-                            Ok(oneshot_rx.await?)
-                        }
-                        None => Err(crate::error::anyhow!("Actor Dropped")),
-                    }
-                })
-            })),
+            actor_id: self.actor_id,
+            caller_fn,
+            test_fn,
         }
     }
 
@@ -143,21 +149,27 @@ impl<A: Actor> Addr<A> {
         A: Handler<T>,
     {
         let weak_tx = Arc::downgrade(&self.tx);
+        let sender_fn = Box::new(move |msg| match weak_tx.upgrade() {
+            Some(tx) => {
+                mpsc::UnboundedSender::clone(&tx).start_send(ActorEvent::Exec(Box::new(
+                    move |actor, ctx| {
+                        Box::pin(async move {
+                            Handler::handle(&mut *actor, ctx, msg).await;
+                        })
+                    },
+                )))?;
+                Ok(())
+            }
+            None => Ok(()),
+        });
+
+        let weak_tx = Arc::downgrade(&self.tx);
+        let test_fn = Box::new(move || weak_tx.strong_count() > 0);
+
         Sender {
-            actor_id: self.actor_id.clone(),
-            sender_fn: Box::new(move |msg| match weak_tx.upgrade() {
-                Some(tx) => {
-                    mpsc::UnboundedSender::clone(&tx).start_send(ActorEvent::Exec(Box::new(
-                        move |actor, ctx| {
-                            Box::pin(async move {
-                                Handler::handle(&mut *actor, ctx, msg).await;
-                            })
-                        },
-                    )))?;
-                    Ok(())
-                }
-                None => Ok(()),
-            }),
+            actor_id: self.actor_id,
+            sender_fn,
+            test_fn,
         }
     }
 
@@ -191,14 +203,11 @@ impl<A> Hash for WeakAddr<A> {
 
 impl<A> WeakAddr<A> {
     pub fn upgrade(&self) -> Option<Addr<A>> {
-        match self.tx.upgrade() {
-            Some(tx) => Some(Addr {
-                actor_id: self.actor_id,
-                tx,
-                rx_exit: self.rx_exit.clone(),
-            }),
-            None => None,
-        }
+        self.tx.upgrade().map(|tx| Addr {
+            actor_id: self.actor_id,
+            tx,
+            rx_exit: self.rx_exit.clone(),
+        })
     }
 }
 

src/caller.rs

@@ -11,19 +11,25 @@ pub(crate) type CallerFn<T> = Box<dyn Fn(T) -> CallerFuture<T> + Send + 'static>
 
 pub(crate) type SenderFn<T> = Box<dyn Fn(T) -> Result<()> + 'static + Send>;
 
+pub(crate) type TestFn = Box<dyn Fn() -> bool + 'static + Send>;
+
 /// Caller of a specific message type
 ///
 /// Like `Sender<T>, Caller has a weak reference to the recipient of the message type, and so will not prevent an actor from stopping if all Addr's have been dropped elsewhere.
 
 pub struct Caller<T: Message> {
     pub actor_id: ActorId,
     pub(crate) caller_fn: Mutex<CallerFn<T>>,
+    pub(crate) test_fn: TestFn,
 }
 
 impl<T: Message> Caller<T> {
     pub fn call(&self, msg: T) -> CallerFuture<T> {
         (self.caller_fn.lock().unwrap())(msg)
     }
+    pub fn can_upgrade(&self) -> bool {
+        (self.test_fn)()
+    }
 }
 
 impl<T: Message<Result = ()>> PartialEq for Caller<T> {
@@ -46,12 +52,16 @@ impl<T: Message<Result = ()>> Hash for Caller<T> {
 pub struct Sender<T: Message> {
     pub actor_id: ActorId,
     pub(crate) sender_fn: SenderFn<T>,
+    pub(crate) test_fn: TestFn,
 }
 
 impl<T: Message<Result = ()>> Sender<T> {
     pub fn send(&self, msg: T) -> Result<()> {
         (self.sender_fn)(msg)
     }
+    pub fn can_upgrade(&self) -> bool {
+        (self.test_fn)()
+    }
 }
 
 impl<T: Message<Result = ()>> PartialEq for Sender<T> {