repository·master·Indexed 18 days ago
https://github.com/restioson/xtraA lightweight, safe, and fast actor framework for Rust, modeled after Actix. It is runtime-agnostic and provides convenience wrappers for executors such as Tokio, async-std, and smol. xtra features an asynchronous Handler interface supporting async/await with &mut self access and includes a custom Actor derive macro. It supports strong and weak addresses, priority-based mailboxes (Default, Priority, and Broadcast), and integration with wasm-bindgen.
Handler interface that supports async/await syntax with &mut self access.When using xtra, you should access all macros through the main xtra crate.
While the macros are implemented in the xtra-macros crate, that crate is an internal implementation detail. The macros-test crate exists solely to verify that the public macro API exposed via the main xtra crate behaves as expected and remains consistent for end-users.
To run an actor using the Tokio runtime, enable the tokio and macros features in your Cargo.toml. You can then use xtra::spawn_tokio to spawn the actor with a specified Mailbox type (e.g., Mailbox::unbounded()).
[dependencies]
extra = { version = "0.6.0", features = ["tokio", "macros"] }// Spawning the actor
let addr = xtra::spawn_tokio(Printer::default(), Mailbox::unbounded());
// Sending a message
addr.send(Print("hello".to_string())).await.expect("Printer should not be dropped");To verify the basic-wasm-bindgen example, you can run the tests using wasm-pack in a headless Firefox environment, or use the provided shell script if you are on a Linux system.
# Using wasm-pack with headless Firefox
wasm-pack test --headless --firefox
# Or on Linux using the provided script
./test.shTo create an actor, define a struct and implement the Actor derive macro. To handle messages, implement the Handler<Message> trait for your actor. The handle method is asynchronous, allowing you to use async/await while maintaining mutable access to the actor's state via &mut self.
use xtra::prelude::*;
#[derive(Default, xtra::Actor)]
struct Printer {
times: usize,
}
struct Print(String);
impl Handler<Print> for Printer {
type Return = ();
async fn handle(&mut self, print: Print, _ctx: &mut Context<Self>) {
self.times += 1;
println!("Printing {}. Printed {} times so far.", print.0, self.times);
}
}A MessageChannel<M, R, Rc> is a communication primitive used to send a specific message type M to any actor capable of handling it, returning a value of type R.
Unlike an Address, which is tied to a specific actor type, a MessageChannel is associated with the message type. This allows you to decouple the sender from the specific identity of the actor, as long as the actor implements Handler<M, Return = R>.
.send(message) to dispatch a message. It returns a SendFuture that resolves to the handler's return value or an error if the actor is disconnected..is_connected() to see if the actor is running, .len() for mailbox size, and .is_empty() to check if the mailbox is empty..join() to get an ActorJoinHandle that resolves when the actor stops..downgrade() a strong channel to a Weak reference or use .as_either() to convert to an Either reference count type.# use xtra::prelude::*;
struct WhatsYourName;
struct Alice;
struct Bob;
impl Actor for Alice {
type Stop = ();
async fn stopped(self) {}
}
impl Actor for Bob {
type Stop = ();
async fn stopped(self) {}
}
impl Handler<WhatsYourName> for Alice {
type Return = &'static str;
async fn handle(&mut self, _: WhatsYourName, _ctx: &mut Context<Self>) -> Self::Return {
"Alice"
}
}
impl Handler<WhatsYourName> for Bob {
type Return = &'static str;
async fn handle(&mut self, _: WhatsYourName, _ctx: &mut Context<Self>) -> Self::Return {
"Bob"
}
}
fn main() {
# #[cfg(feature = "smol")]
smol::block_on(async {
let alice = xtra::spawn_smol(Alice, Mailbox::unbounded());
let bob = xtra::spawn_smol(Bob, Mailbox::unbounded());
let channels = [
MessageChannel::new(alice),
MessageChannel::new(bob)
];
let name = ["Alice", "Bob"];
for (channel, name) in channels.iter().zip(&name) {
assert_eq!(*name, channel.send(WhatsYourName).await.unwrap());
}
})
}# use xtra::prelude::*;
// ... (see full example in content) ...
let channels = [
MessageChannel::new(alice),
MessageChannel::new(bob)
];
assert_eq!(*name, channel.send(WhatsYourName).await.unwrap());An Address<A, Rc> is a reference to an actor used to send messages. By default, addresses are Strong, meaning they prevent the actor from being dropped as long as at least one strong address exists.
If you want to hold a reference to an actor without preventing it from being shut down and dropped, use a WeakAddress<A>. You can create one by calling Address::downgrade() on a strong address.
To use a WeakAddress, you must attempt to upgrade it back to a strong Address using try_upgrade(). This returns None if the actor has already been dropped.
// Creating a weak address from a strong one
let weak_addr = strong_addr.downgrade();
// Attempting to use it later
if let Some(strong_addr) = weak_addr.try_upgrade() {
strong_addr.send(Message).await;
}The xtra actor system follows a pattern where an Actor holds state and lifecycle logic, while Handler implementations define the behavior for specific message types.
started, stopped).handle.spawn_smol) to start the actor and receive an Address.# use xtra::prelude::*;
# struct MyActor;
# impl Actor for MyActor { type Stop = (); async fn stopped(self) {} }
# struct Msg;
impl Handler<Msg> for MyActor {
type Return = u32;
async fn handle(&mut self, _message: Msg, _ctx: &mut Context<Self>) -> u32 {
20
}
}
// Usage pattern:
// let addr = xtra::spawn_smol(MyActor, Mailbox::unbounded());
// let result = addr.send(Msg).await;A SendFuture represents the state of sending a message to an actor. It handles the lifecycle of queuing a message into an actor's mailbox and managing the response.
By default, awaiting a SendFuture resolves directly to the return value of the actor's handler (Handler::Return).
If the actor's mailbox is bounded, the SendFuture will yield Pending until the message is successfully queued. This allows the actor to exercise backpressure on its users.
You can call .detach() on a SendFuture to change its behavior. A detached future resolves once the message is successfully queued, returning a Receiver<R>. This Receiver is itself a future that resolves to the handler's return value. This pattern allows your current task to continue immediately after the message is queued, while the actor processes the message asynchronously.
You can set the priority of a message using .priority(new_priority: u32) before the future is polled.
Note: Calling .priority() after the future has already been polled will cause a panic.
// Example of detaching a SendFuture to handle the response separately
let receiver = send_future.detach().await?;
// ... do other work ...
let result = receiver.await?;In the xtra actor model, an Address<A> and a Mailbox<A> function as a Multi-Producer Multi-Consumer (MPMC) channel.
Address<A>: The sending end. Messages sent to an Address are queued for delivery.Mailbox<A>: The receiving end. It is the counterpart to the Address and is used by an actor to retrieve incoming messages.You can create these pairs using either Mailbox::bounded(capacity) for back-pressure or Mailbox::unbounded() for unlimited capacity (use with caution due to potential memory growth).
// Example of creating a bounded mailbox/address pair
let capacity = 100;
let (address, mailbox) = Mailbox::<MyMessage>::bounded(capacity);
// Sending a message
address.send(MyMessage::Hello);
// Receiving a message
let msg = mailbox.next().await;A DispatchFuture is a state machine that manages the transition from a raw Message to the execution of an actor's handler.
ActorMessage, a mutable reference to the actor, and the Mailbox.Running state by invoking the appropriate handler logic (e.g., ToOneActor, ToAllActors, or Shutdown). It wraps the resulting handler future in a BoxFuture.Span (if instrumentation is enabled).Poll::Ready(ControlFlow<()>) once the handler completes.Internally, an actor's mailbox is divided into three distinct queues, each with its own capacity and backpressure behavior:
Address::broadcast will wait for the slowest actor to process the message.Note that the actor's mailbox capacity applies to each queue individually. An actor can have cap messages in every mailbox before it is considered full.