Struct Context 

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

An actor execution context.

Implementations

impl<A> Context<A>

pub fn address(&self) -> Addr<A>

Returns the address of the actor.

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(())
    }

pub fn actor_id(&self) -> ActorId

Returns the id of the actor.

Examples found in repository

examples/supervisor_clear_interval.rs (line 41)

    async fn handle(&mut self, ctx: &mut Context<Self>, _msg: Ping) {
        let now = Instant::now();
        let delta = (now - self.last_ping).as_millis();
        self.last_ping = now;
        println!("PingTimer:: Ping {} {:?}", ctx.actor_id(), delta);
    }

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

Stop the actor.

Examples found in repository

examples/main.rs (line 22)

    async fn handle(&mut self, ctx: &mut Context<Self>, _msg: Die) {
        // Stop the actor without error
        ctx.stop(None);
    }

examples/supervisor_clear_interval.rs (line 51)

    async fn handle(&mut self, ctx: &mut Context<Self>, _msg: Halt) {
        println!("PingTimer:: received Halt");
        ctx.stop(None);
        println!("PingTimer:: stopped");
    }

examples/supervisor_clear_send_later.rs (line 37)

    async fn handle(&mut self, ctx: &mut Context<Self>, _msg: Halt) {
        println!("PingLater:: received Halt");
        ctx.stop(None);
        println!("PingLater:: stopped");
    }

pub fn abort_intervals(&mut self)

pub fn abort_streams(&mut self)

pub fn add_stream<S>(&mut self, stream: S)

where S: Stream + Unpin + Send + 'static, S::Item: 'static + Send, A: StreamHandler<S::Item>,

Create a stream handler for the actor.

Examples

use xactor::*;
use futures::stream;
use std::time::Duration;

#[message(result = "i32")]
struct GetSum;

#[derive(Default)]
struct MyActor(i32);

#[async_trait::async_trait]
impl StreamHandler<i32> for MyActor {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: i32) {
        self.0 += msg;
    }

    async fn started(&mut self, _ctx: &mut Context<Self>) {
        println!("stream started");
    }

    async fn finished(&mut self, _ctx: &mut Context<Self>) {
        println!("stream finished");
    }
}

#[async_trait::async_trait]
impl Handler<GetSum> for MyActor {
    async fn handle(&mut self, _ctx: &mut Context<Self>, _msg: GetSum) -> i32 {
        self.0
    }
}

#[async_trait::async_trait]
impl Actor for MyActor {
    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        let values = (0..100).collect::<Vec<_>>();
        ctx.add_stream(stream::iter(values));
        Ok(())
    }
}

#[xactor::main]
async fn main() -> Result<()> {
    let mut addr = MyActor::start_default().await?;
    sleep(Duration::from_secs(1)).await; // Wait for the stream to complete
    let res = addr.call(GetSum).await?;
    assert_eq!(res, (0..100).sum::<i32>());
    Ok(())
}

pub fn send_later<T>(&mut self, msg: T, after: Duration)

where A: Handler<T>, T: Message<Result = ()>,

Sends the message msg to self after a specified period of time.

We use Sender instead of Addr so that the interval doesn’t keep reference to address and prevent the actor from being dropped and stopped

Examples found in repository

examples/supervisor_clear_send_later.rs (line 10)

    async fn started(&mut self, ctx: &mut Context<Self>) -> xactor::Result<()> {
        ctx.send_later(Ping("after halt"), Duration::from_millis(1_500));

        Ok(())
    }

examples/main.rs (line 13)

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        // Send the Die message 3 seconds later
        ctx.send_later(Die, Duration::from_secs(3));
        Ok(())
    }

pub fn send_interval_with<T, F>(&mut self, f: F, dur: Duration)

where A: Handler<T>, F: Fn() -> T + Sync + Send + 'static, T: Message<Result = ()>,

Sends the message to self, at a specified fixed interval. The message is created each time using a closure f.

pub fn send_interval<T>(&mut self, msg: T, dur: Duration)

where A: Handler<T>, T: Message<Result = ()> + Clone + Sync,

Sends the message msg to self, at a specified fixed interval.

Examples found in repository

examples/supervisor_clear_interval.rs (line 21)

    async fn started(&mut self, ctx: &mut Context<Self>) -> xactor::Result<()> {
        println!("PingTimer:: started()");
        ctx.send_interval(Ping, Duration::from_millis(1000));
        Ok(())
    }

examples/subscriber.rs (line 135)

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        // Send broadcast message to self every 2 seconds
        println!("Message Producer Started");
        ctx.send_interval(Broadcast, Duration::from_secs(2));
        Ok(())
    }

pub async fn subscribe<T: Message<Result = ()>>(&self) -> Result<()>

where A: Handler<T>,

Subscribes to a message of a specified type.

pub async fn unsubscribe<T: Message<Result = ()>>(&self) -> Result<()>

Unsubscribe to a message of a specified type.