Messages priority & inbox unification

Cargo.toml

@@ -13,16 +13,15 @@ categories = ["asynchronous", "concurrency"]
 rust-version = "1.56.0"
 
 [dependencies]
-async-trait = "0.1.36"
-barrage = "0.2.2"
+async-trait = "0.1.56"
 catty = "0.1.5"
-flume = { version = "0.10.13", default-features = false, features = ["async"] }
 futures-core = { version = "0.3.21", default-features = false, features = ["alloc"] }
 futures-sink = { version = "0.3.21", default-features = false }
 futures-util = { version = "0.3.21", default-features = false, features = ["sink", "alloc"] }
 pollster = "0.2"
 pin-project-lite = "0.2.9"
 event-listener = "2.4.0"
+spin = { version = "0.9.3", default-features = false, features = ["spin_mutex"] }
 
 # Feature `timing`
 futures-timer = { version = "3.0", optional = true, default-features = false }
@@ -41,7 +40,7 @@ wasm-bindgen = { version = "0.2", optional = true, default-features = false }
 wasm-bindgen-futures = { version = "0.4", optional = true, default-features = false }
 
 # Feature `with-tracing-0_1`
-tracing = { version = "0.1", optional = true, default-features = false }
+tracing = { version = "0.1.35", optional = true, default-features = false }
 
 [dev-dependencies]
 rand = "0.8"
@@ -96,7 +95,7 @@ name = "backpressure"
 required-features = ["with-tokio-1"]
 
 [[example]]
-name = "address_is_sink"
+name = "address_sink"
 required-features = ["with-tokio-1", "tokio/full"]
 
 [[example]]

README.md

@@ -5,7 +5,7 @@ For better ergonomics with xtra, try the [spaad](https://crates.io/crates/spaad)
 
 ## Features
 - Safe: there is no unsafe code in xtra.
-- Tiny: xtra is less than 2kloc.
+- Tiny: xtra is around 2kloc.
 - Lightweight: xtra has few dependencies, most of which are lightweight (except `futures`).
 - Asynchronous and synchronous message handlers.
 - Simple asynchronous message handling interface which allows `async`/`await` syntax even when borrowing `self`.
@@ -32,7 +32,11 @@ impl Printer {
     }
 }
 
-impl Actor for Printer {}
+#[async_trait]
+impl Actor for Printer {
+    type Stop = ();
+    async fn stopped(self) {}
+}
 
 struct Print(String);
 
@@ -41,7 +45,7 @@ impl Handler<Print> for Printer {
     type Return = ();
 
     async fn handle(&mut self, print: Print, _ctx: &mut Context<Self>) {
-        self.times += 1; // no ActorFuture or anything just to access `self`
+        self.times += 1;
         println!("Printing {}. Printed {} times so far.", print.0, self.times);
     }
 }

examples/address_is_sink.rs → examples/address_sink.rs

@@ -46,7 +46,7 @@ async fn main() {
     repeat(10)
         .take(4)
         .map(|number| Ok(Add(number)))
-        .forward(addr.clone())
+        .forward(addr.clone().into_sink())
         .await
         .unwrap();
 

examples/basic_wasm_bindgen/Cargo.toml

@@ -8,7 +8,7 @@ edition = "2018"
 crate-type = ["cdylib", "rlib"]
 
 [dependencies]
-wasm-bindgen = { version = "0.2.68", default-features = false }
+wasm-bindgen = { version = "0.2.81", default-features = false }
 wasm-bindgen-futures = { version = "0.4.13", default-features = false }
 xtra = { path = "../..", features = ["with-wasm_bindgen-0_2"] }
 async-trait = "0.1"

examples/crude_bench.rs

@@ -1,12 +1,10 @@
-use std::time::{Duration, Instant};
-
-use futures_core::future::BoxFuture;
+use futures_util::FutureExt;
 use std::future::Future;
+use std::time::{Duration, Instant};
 use xtra::prelude::*;
 use xtra::spawn::Tokio;
-use xtra::NameableSending;
-use xtra::Receiver;
 use xtra::SendFuture;
+use xtra::{ActorErasedSending, NameableSending};
 
 struct Counter {
     count: usize,
@@ -94,7 +92,7 @@ impl Handler<TimeReturn> for ReturnTimer {
     }
 }
 
-const COUNT: usize = 50_000_000; // May take a while on some machines
+const COUNT: usize = 10_000_000; // May take a while on some machines
 
 async fn do_address_benchmark<R>(
     name: &str,
@@ -108,7 +106,7 @@ async fn do_address_benchmark<R>(
 
     // rounding overflow
     for _ in 0..COUNT {
-        let _ = f(&addr).await;
+        let _ = f(&addr).now_or_never();
     }
 
     // awaiting on GetCount will make sure all previous messages are processed first BUT introduces
@@ -149,13 +147,11 @@ async fn do_parallel_address_benchmark<R>(
 
 async fn do_channel_benchmark<M, RM>(
     name: &str,
-    f: impl Fn(
-        &dyn MessageChannel<M, Return = ()>,
-    ) -> SendFuture<(), BoxFuture<'static, Receiver<()>>, RM>,
+    f: impl Fn(&dyn MessageChannel<M, Return = ()>) -> SendFuture<(), ActorErasedSending<()>, RM>,
 ) where
     Counter: Handler<M, Return = ()> + Send,
     M: Send + 'static,
-    SendFuture<(), BoxFuture<'static, Receiver<()>>, RM>: Future,
+    SendFuture<(), ActorErasedSending<()>, RM>: Future,
 {
     let addr = Counter { count: 0 }.create(None).spawn(&mut Tokio::Global);
     let chan = &addr as &dyn MessageChannel<M, Return = ()>;

examples/interleaved_messages.rs

@@ -43,8 +43,7 @@ impl Handler<Initialized> for ActorB {
     async fn handle(&mut self, m: Initialized, ctx: &mut Context<Self>) {
         println!("ActorB: Initialized");
         let actor_a = m.0;
-        let send = actor_a.send(Hello);
-        ctx.join(self, send).await.unwrap();
+        ctx.join(self, actor_a.send(Hello)).await.unwrap();
     }
 }
 

examples/send_interval.rs

@@ -30,8 +30,7 @@ impl Handler<Greet> for Greeter {
 #[tokio::main]
 async fn main() {
     let addr = Greeter::default().create(None).spawn(&mut Tokio::Global);
-
-    greeter_stream(500).forward(addr).await.unwrap();
+    greeter_stream(500).forward(addr.into_sink()).await.unwrap();
 }
 
 fn greeter_stream(delay: u64) -> impl Stream<Item = Result<Greet, Disconnected>> {