tokio executor

Cargo.toml

@@ -13,6 +13,10 @@ keywords = ["actors", "actor-model", "async", "cqrs", "event_sourcing"]
 [badges]
 travis-ci = { repository = "riker-rs/riker" }
 
+[features]
+default = []
+tokio_executor = ["tokio"]
+
 [dependencies]
 riker-macros = { path = "riker-macros", version = "0.2.0" }
 chrono = "0.4"
@@ -27,8 +31,12 @@ slog-stdlog = "4.0"
 slog-scope = "4.3.0"
 num_cpus = "1.13.0"
 dashmap = "3"
+tokio = { version = "^1", features = ["rt-multi-thread", "macros", "time"], optional = true }
 
 
 [dev-dependencies]
-riker-testkit = "0.1.0"
 log = "0.4"
+
+#[dev-dependencies.riker-testkit]
+#git = "https://github.com/mankinskin/riker-testkit"
+#branch = "tokio_executor"

riker-macros/Cargo.toml

@@ -13,10 +13,18 @@ keywords = ["actors", "actor-model", "async", "cqrs", "event_sourcing"]
 [lib]
 proc-macro = true
 
+[features]
+default = []
+
 [dependencies]
 syn = { version ="1.0", features = ["parsing", "full", "extra-traits", "proc-macro"] }
 quote = "1.0"
 proc-macro2 = "1.0"
+tokio = { version = "^1", features = ["rt-multi-thread", "macros", "time"], optional = true }
 
 [dev-dependencies]
 riker = { path = ".." }
+
+[dev-dependencies.riker-testkit]
+git = "https://github.com/mankinskin/riker-testkit"
+branch = "tokio_executor"

riker-macros/tests/macro.rs

@@ -33,7 +33,7 @@ impl Receive<String> for NewActor {
     }
 }
 
-#[test]
+#[riker_testkit::test]
 fn run_derived_actor() {
     let sys = ActorSystem::new().unwrap();
 
@@ -45,6 +45,9 @@ fn run_derived_actor() {
     // wait until all direct children of the user root are terminated
     while sys.user_root().has_children() {
         // in order to lower cpu usage, sleep here
+        #[cfg(feature = "tokio_executor")]
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        #[cfg(not(feature = "tokio_executor"))]
         std::thread::sleep(std::time::Duration::from_millis(50));
     }
 }
@@ -80,7 +83,7 @@ impl<A: Send + 'static, B: Send + 'static> Receive<String> for GenericActor<A, B
     }
 }
 
-#[test]
+#[riker_testkit::test]
 fn run_derived_generic_actor() {
     let sys = ActorSystem::new().unwrap();
 
@@ -92,6 +95,9 @@ fn run_derived_generic_actor() {
     // wait until all direct children of the user root are terminated
     while sys.user_root().has_children() {
         // in order to lower cpu usage, sleep here
+        #[cfg(feature = "tokio_executor")]
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        #[cfg(not(feature = "tokio_executor"))]
         std::thread::sleep(std::time::Duration::from_millis(50));
     }
 }
@@ -131,7 +137,7 @@ impl Receive<Message<String>> for GenericMsgActor {
     }
 }
 
-#[test]
+#[riker_testkit::test]
 fn run_generic_message_actor() {
     let sys = ActorSystem::new().unwrap();
 
@@ -145,6 +151,9 @@ fn run_generic_message_actor() {
     // wait until all direct children of the user root are terminated
     while sys.user_root().has_children() {
         // in order to lower cpu usage, sleep here
+        #[cfg(feature = "tokio_executor")]
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        #[cfg(not(feature = "tokio_executor"))]
         std::thread::sleep(std::time::Duration::from_millis(50));
     }
 }
@@ -202,7 +211,7 @@ impl Receive<test_mod::Message> for PathMsgActor {
     }
 }
 
-#[test]
+#[riker_testkit::test]
 fn run_path_message_actor() {
     let sys = ActorSystem::new().unwrap();
 
@@ -219,6 +228,9 @@ fn run_path_message_actor() {
     // wait until all direct children of the user root are terminated
     while sys.user_root().has_children() {
         // in order to lower cpu usage, sleep here
+        #[cfg(feature = "tokio_executor")]
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        #[cfg(not(feature = "tokio_executor"))]
         std::thread::sleep(std::time::Duration::from_millis(50));
     }
 }

src/actor.rs

@@ -210,58 +210,63 @@ impl<A: Actor + ?Sized> Actor for Box<A> {
 /// attribute macro and implemented for each message type to receive.
 ///
 /// # Examples
-///
-/// ```
-/// # use riker::actors::*;
-///
-/// #[derive(Clone, Debug)]
-/// pub struct Foo;
-/// #[derive(Clone, Debug)]
-/// pub struct Bar;
-/// #[actor(Foo, Bar)] // <-- set our actor to receive Foo and Bar types
-/// #[derive(Default)]
-/// struct MyActor;
-///
-/// impl Actor for MyActor {
-///     type Msg = MyActorMsg; // <-- MyActorMsg is provided for us
-///
-///     fn recv(&mut self,
-///                 ctx: &Context<Self::Msg>,
-///                 msg: Self::Msg,
-///                 sender: Sender) {
-///         self.receive(ctx, msg, sender); // <-- call the respective implementation
-///     }
-/// }
-///
-/// impl Receive<Foo> for MyActor {
-///     type Msg = MyActorMsg;
-///
-///     fn receive(&mut self,
-///                 ctx: &Context<Self::Msg>,
-///                 msg: Foo, // <-- receive Foo
-///                 sender: Sender) {
-///         println!("Received a Foo");
-///     }
-/// }
-///
-/// impl Receive<Bar> for MyActor {
-///     type Msg = MyActorMsg;
-///
-///     fn receive(&mut self,
-///                 ctx: &Context<Self::Msg>,
-///                 msg: Bar, // <-- receive Bar
-///                 sender: Sender) {
-///         println!("Received a Bar");
-///     }
-/// }
-///
-/// // main
-/// let sys = ActorSystem::new().unwrap();
-/// let actor = sys.actor_of::<MyActor>("my-actor").unwrap();
-///
-/// actor.tell(Foo, None);
-/// actor.tell(Bar, None);
-/// ```
+#[cfg_attr(
+    not(feature = "tokio_executor"),
+    doc = r##"
+```
+# use riker::actors::*;
+
+#[derive(Clone, Debug)]
+pub struct Foo;
+#[derive(Clone, Debug)]
+pub struct Bar;
+#[actor(Foo, Bar)] // <-- set our actor to receive Foo and Bar types
+#[derive(Default)]
+struct MyActor;
+
+impl Actor for MyActor {
+    type Msg = MyActorMsg; // <-- MyActorMsg is provided for us
+
+    fn recv(&mut self,
+                ctx: &Context<Self::Msg>,
+                msg: Self::Msg,
+                sender: Sender) {
+        self.receive(ctx, msg, sender); // <-- call the respective implementation
+    }
+}
+
+impl Receive<Foo> for MyActor {
+    type Msg = MyActorMsg;
+
+    fn receive(&mut self,
+                ctx: &Context<Self::Msg>,
+                msg: Foo, // <-- receive Foo
+                sender: Sender) {
+        println!("Received a Foo");
+    }
+}
+
+impl Receive<Bar> for MyActor {
+    type Msg = MyActorMsg;
+
+    fn receive(&mut self,
+                ctx: &Context<Self::Msg>,
+                msg: Bar, // <-- receive Bar
+                sender: Sender) {
+        println!("Received a Bar");
+    }
+}
+
+fn main() {
+    let sys = ActorSystem::new().unwrap();
+    let actor = sys.actor_of::<MyActor>("my-actor").unwrap();
+
+    actor.tell(Foo, None);
+    actor.tell(Bar, None);
+}
+```
+"##
+)]
 pub trait Receive<Msg: Message> {
     type Msg: Message;
 

src/actor/actor_cell.rs

@@ -10,11 +10,12 @@ use std::{
 
 use chrono::prelude::*;
 use dashmap::DashMap;
-use futures::{future::RemoteHandle, task::SpawnError, Future};
+use futures::Future;
 use uuid::Uuid;
 
 use crate::{
     actor::{props::ActorFactory, *},
+    executor::TaskHandle,
     kernel::{
         kernel_ref::{dispatch, dispatch_any, KernelRef},
         mailbox::{AnyEnqueueError, AnySender, MailboxSender},
@@ -525,7 +526,10 @@ impl<Msg> Run for Context<Msg>
 where
     Msg: Message,
 {
-    fn run<Fut>(&self, future: Fut) -> Result<RemoteHandle<<Fut as Future>::Output>, SpawnError>
+    fn run<Fut>(
+        &self,
+        future: Fut,
+    ) -> Result<TaskHandle<<Fut as Future>::Output>, Box<dyn std::error::Error>>
     where
         Fut: Future + Send + 'static,
         <Fut as Future>::Output: Send,