Struct Addr 

pub struct Addr<A> { /* private fields */ }

The address of an actor.

When all references to Addr<A> are dropped, the actor ends. You can use Clone trait to create multiple copies of Addr<A>.

Implementations

impl<A> Addr<A>

pub fn downgrade(&self) -> WeakAddr<A>

impl<A: Actor> Addr<A>

pub fn actor_id(&self) -> ActorId

Returns the id of the actor.

pub fn stop(&mut self, err: Option<Error>) -> Result<()>

Stop the actor.

pub async fn call<T: Message>(&self, msg: T) -> Result<T::Result>

where A: Handler<T>,

Send a message msg to the actor and wait for the return value.

Examples found in repository

examples/ping.rs (line 30)

async fn main() -> Result<()> {
    // start new actor
    let addr = MyActor { count: 10 }.start().await?;

    // send message and get future for result
    let res = addr.call(Ping(10)).await?;
    println!("RESULT: {}", res == 20);

    Ok(())
}

pub fn send<T: Message<Result = ()>>(&self, msg: T) -> Result<()>

where A: Handler<T>,

Send a message msg to the actor without waiting for the return value.

Examples found in repository

examples/subscriber.rs (line 36)

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        println!("Subscriber Parent Started");
        let _ = ctx.address().send(InitializeChildSubscribers);
        Ok(())
    }
}

#[message]
struct InitializeChildSubscribers;

#[async_trait::async_trait]
impl Handler<InitializeChildSubscribers> for SubscriberParent {
    async fn handle(&mut self, _ctx: &mut Context<Self>, _msg: InitializeChildSubscribers) {
        let message_producer_addr = self.message_producer.clone();
        let dummy_ids: Vec<i32> = vec![1, 2, 3, 4, 5];
        let children_unstarted_actors_vec = dummy_ids.into_iter().map(move |id| {
            let id = id.clone();
            let addr = message_producer_addr.clone();

            Subscriber::new(id, addr)
        });

        let children_addr_vec = children_unstarted_actors_vec
            .into_iter()
            .map(|actor| async { actor.start().await.unwrap() });

        let children_addr_vec = futures::future::join_all(children_addr_vec).await;

        self.children_subscribers = children_addr_vec;
    }
}

#[message]
struct ClearChildSubscribers;

#[async_trait::async_trait]
impl Handler<ClearChildSubscribers> for SubscriberParent {
    async fn handle(&mut self, _ctx: &mut Context<Self>, _msg: ClearChildSubscribers) {
        self.children_subscribers = Vec::new();
    }
}

// (Child) Subscriber - B

struct Subscriber {
    id: i32,
    message_producer_addr: Addr<MessageProducer>,
}

impl Subscriber {
    fn new(id: i32, message_producer_addr: Addr<MessageProducer>) -> Subscriber {
        Subscriber {
            id,
            message_producer_addr,
        }
    }
}

#[async_trait::async_trait]
impl Actor for Subscriber {
    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        // Send subscription request message to the Message Producer
        println!("Child Subscriber Started - id {:?}", self.id);
        let self_sender = ctx.address().sender();
        let _ = self.message_producer_addr.send(SubscribeToProducer {
            sender: self_sender,
        });
        Ok(())
    }

examples/supervisor_clear_send_later.rs (line 53)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let service_supervisor = xactor::Supervisor::start(PingLater::default).await?;
    let service_addr = service_supervisor.clone();

    let supervisor_task = xactor::spawn(async {
        service_supervisor.wait_for_stop().await;
    });

    let send_ping = async {
        println!("  main  :: sending Ping");
        service_addr.send(Ping("before halt")).unwrap();
    };

    let send_halt = async {
        xactor::sleep(Duration::from_millis(1_000)).await;
        println!("  main  :: sending Halt");
        service_addr.send(Halt).unwrap();
    };

    let _ = futures::join!(supervisor_task, send_halt, send_ping);

    Ok(())
}

examples/supervisor_clear_interval.rs (line 79)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let service_supervisor = xactor::Supervisor::start(PingTimer::default).await?;
    let service_addr = service_supervisor.clone();

    let supervisor_task = xactor::spawn(async {
        service_supervisor.wait_for_stop().await;
    });

    let send_halt = async {
        xactor::sleep(Duration::from_millis(5_200)).await;
        println!("  main  :: sending Halt");
        service_addr.send(Halt).unwrap();
    };

    let _ = futures::join!(supervisor_task, send_halt);
    // run this to see that the interval is not properly stopped if the ctx is stopped
    // futures::join!(supervisor_task, send_panic); // there is no panic recovery

    Ok(())
}

pub fn caller<T: Message>(&self) -> Caller<T>

where A: Handler<T>,

Create a Caller<T> for a specific message type

pub fn sender<T: Message<Result = ()>>(&self) -> Sender<T>

where A: Handler<T>,

Create a Sender<T> for a specific message type

Examples found in repository

examples/subscriber.rs (line 97)

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        // Send subscription request message to the Message Producer
        println!("Child Subscriber Started - id {:?}", self.id);
        let self_sender = ctx.address().sender();
        let _ = self.message_producer_addr.send(SubscribeToProducer {
            sender: self_sender,
        });
        Ok(())
    }

pub async fn wait_for_stop(self)

Wait for an actor to finish, and if the actor has finished, the function returns immediately.

Examples found in repository

examples/main.rs (line 30)

async fn main() -> Result<()> {
    // Exit the program after 3 seconds
    let addr = MyActor.start().await?;
    addr.wait_for_stop().await;
    Ok(())
}

examples/subscriber.rs (line 12)

async fn main() -> std::io::Result<()> {
    let parent_addr = SubscriberParent::new().await.start().await.unwrap();
    parent_addr.wait_for_stop().await;
    Ok(())
}

examples/supervisor_clear_send_later.rs (line 48)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let service_supervisor = xactor::Supervisor::start(PingLater::default).await?;
    let service_addr = service_supervisor.clone();

    let supervisor_task = xactor::spawn(async {
        service_supervisor.wait_for_stop().await;
    });

    let send_ping = async {
        println!("  main  :: sending Ping");
        service_addr.send(Ping("before halt")).unwrap();
    };

    let send_halt = async {
        xactor::sleep(Duration::from_millis(1_000)).await;
        println!("  main  :: sending Halt");
        service_addr.send(Halt).unwrap();
    };

    let _ = futures::join!(supervisor_task, send_halt, send_ping);

    Ok(())
}

examples/supervisor_clear_interval.rs (line 73)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let service_supervisor = xactor::Supervisor::start(PingTimer::default).await?;
    let service_addr = service_supervisor.clone();

    let supervisor_task = xactor::spawn(async {
        service_supervisor.wait_for_stop().await;
    });

    let send_halt = async {
        xactor::sleep(Duration::from_millis(5_200)).await;
        println!("  main  :: sending Halt");
        service_addr.send(Halt).unwrap();
    };

    let _ = futures::join!(supervisor_task, send_halt);
    // run this to see that the interval is not properly stopped if the ctx is stopped
    // futures::join!(supervisor_task, send_panic); // there is no panic recovery

    Ok(())
}

impl<T: Message<Result = ()> + Clone> Addr<Broker<T>>

pub fn publish(&mut self, msg: T) -> Result<()>

Publishes a message of the specified type.

Trait Implementations

impl<A> Clone for Addr<A>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<A> Hash for Addr<A>

fn hash<H: Hasher>(&self, state: &mut H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl<A> PartialEq for Addr<A>

fn eq(&self, other: &Self) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.