Struct actix::prelude::Addr

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

The address of an actor.

Implementations

impl<A: Actor> Addr<A>

pub fn new(tx: AddressSender<A>) -> Addr<A>

pub fn connected(&self) -> bool

Returns whether the actor is still alive.

pub fn do_send<M>(&self, msg: M)

where M: Message + Send, M::Result: Send, A: Handler<M>, A::Context: ToEnvelope<A, M>,

Sends a message unconditionally, ignoring any potential errors.

The message is always queued, even if the mailbox for the receiver is full. If the mailbox is closed, the message is silently dropped.

Examples found in repository

examples/weak_addr.rs (line 29)

    fn send_tick(&mut self, _ctx: &mut Context<Self>) {
        for client in self.clients.iter() {
            if let Some(client) = client.upgrade() {
                client.do_send(TimePing(Instant::now()));
                println!("⏰ sent ping to client {:?}", client);
            } else {
                println!("⏰ client can no longer be upgraded");
            }
        }

        // gc
        self.clients = self
            .clients
            .drain(..)
            .filter(|c| c.upgrade().is_some())
            .collect();
        println!("⏰ service has {} clients", self.clients.len());
    }

pub fn try_send<M>(&self, msg: M) -> Result<(), SendError<M>>

where M: Message + Send + 'static, M::Result: Send, A: Handler<M>, A::Context: ToEnvelope<A, M>,

Tries to send a message.

This method fails if actor’s mailbox is full or closed. This method registers the current task in the receiver’s queue.

pub fn send<M>(&self, msg: M) -> Request<A, M>

where M: Message + Send + 'static, M::Result: Send, A: Handler<M>, A::Context: ToEnvelope<A, M>,

Sends an asynchronous message and waits for a response.

The communication channel to the actor is bounded. If the returned request future gets dropped, the message is cancelled.

Examples found in repository

examples/mock.rs (line 23)

    fn handle(&mut self, _msg: Question, _ctx: &mut Context<Self>) -> Self::Result {
        let act_addr = AnswerActor::from_registry();
        let request = act_addr.send(Question {});
        Box::pin(async move { request.await.unwrap() })
    }

examples/weak_addr.rs (line 89)

    fn started(&mut self, ctx: &mut Self::Context) {
        println!("🐰 starting Client");
        TimeService::from_registry()
            .send(RegisterForTime(ctx.address().downgrade()))
            .into_actor(self)
            .then(|_, _slf, _| fut::ready(()))
            .spawn(ctx);
    }

examples/weak_recipient.rs (line 89)

    fn started(&mut self, ctx: &mut Self::Context) {
        println!("🐰 starting ClientA");
        TimeService::from_registry()
            .send(RegisterForTime(ctx.address().downgrade().recipient()))
            .into_actor(self)
            .then(|_, _slf, _| fut::ready(()))
            .spawn(ctx);
    }

    fn stopping(&mut self, _ctx: &mut Self::Context) -> Running {
        println!("🐰 stopping ClientA");
        Running::Stop
    }

    fn stopped(&mut self, _ctx: &mut Self::Context) {
        println!("🐰 stopped ClientA");
    }
}

impl Handler<TimePing> for ClientA {
    type Result = ();

    fn handle(&mut self, msg: TimePing, _ctx: &mut Self::Context) -> Self::Result {
        println!("🐰 ClientA received ping: {:?}", msg.0);
    }
}

#[derive(Debug, Default)]
pub struct ClientB;

impl Actor for ClientB {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        println!("🐇 starting ClientB");
        TimeService::from_registry()
            .send(RegisterForTime(ctx.address().downgrade().recipient()))
            .into_actor(self)
            .then(|_, _slf, _| fut::ready(()))
            .spawn(ctx);
    }

examples/ring.rs (line 100)

fn main() -> io::Result<()> {
    let sys = System::new();

    let (n_nodes, n_rounds) = parse_args();

    let now = SystemTime::now();

    sys.block_on(async {
        println!("Setting up {} nodes", n_nodes);
        let limit = n_nodes * n_rounds;

        let node = Node::create(move |ctx| {
            let first_addr = ctx.address();

            let mut prev_addr = Node {
                id: 1,
                limit,
                next: first_addr.recipient(),
            }
            .start();

            for id in 2..n_nodes {
                prev_addr = Node {
                    id,
                    limit,
                    next: prev_addr.recipient(),
                }
                .start();
            }

            Node {
                id: n_nodes,
                limit,
                next: prev_addr.recipient(),
            }
        });

        println!(
            "Sending start message and waiting for termination after {} messages...",
            limit
        );

        node.send(Payload(1)).await.unwrap();
    });

    sys.run().unwrap();

    match now.elapsed() {
        Ok(elapsed) => println!(
            "Time taken: {}.{:06} seconds ({} msg/second)",
            elapsed.as_secs(),
            elapsed.subsec_micros(),
            (n_nodes * n_rounds * 1000000) as u128 / elapsed.as_micros()
        ),
        Err(e) => println!("An error occurred: {:?}", e),
    }

    Ok(())
}

pub fn recipient<M>(self) -> Recipient<M>

where A: Handler<M>, A::Context: ToEnvelope<A, M>, M: Message + Send + 'static, M::Result: Send,

Returns the Recipient for a specific message type.

Examples found in repository

examples/ring.rs (line 75)

fn main() -> io::Result<()> {
    let sys = System::new();

    let (n_nodes, n_rounds) = parse_args();

    let now = SystemTime::now();

    sys.block_on(async {
        println!("Setting up {} nodes", n_nodes);
        let limit = n_nodes * n_rounds;

        let node = Node::create(move |ctx| {
            let first_addr = ctx.address();

            let mut prev_addr = Node {
                id: 1,
                limit,
                next: first_addr.recipient(),
            }
            .start();

            for id in 2..n_nodes {
                prev_addr = Node {
                    id,
                    limit,
                    next: prev_addr.recipient(),
                }
                .start();
            }

            Node {
                id: n_nodes,
                limit,
                next: prev_addr.recipient(),
            }
        });

        println!(
            "Sending start message and waiting for termination after {} messages...",
            limit
        );

        node.send(Payload(1)).await.unwrap();
    });

    sys.run().unwrap();

    match now.elapsed() {
        Ok(elapsed) => println!(
            "Time taken: {}.{:06} seconds ({} msg/second)",
            elapsed.as_secs(),
            elapsed.subsec_micros(),
            (n_nodes * n_rounds * 1000000) as u128 / elapsed.as_micros()
        ),
        Err(e) => println!("An error occurred: {:?}", e),
    }

    Ok(())
}

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

Returns a downgraded WeakAddr.

Examples found in repository

examples/weak_addr.rs (line 89)

    fn started(&mut self, ctx: &mut Self::Context) {
        println!("🐰 starting Client");
        TimeService::from_registry()
            .send(RegisterForTime(ctx.address().downgrade()))
            .into_actor(self)
            .then(|_, _slf, _| fut::ready(()))
            .spawn(ctx);
    }

examples/weak_recipient.rs (line 89)

    fn started(&mut self, ctx: &mut Self::Context) {
        println!("🐰 starting ClientA");
        TimeService::from_registry()
            .send(RegisterForTime(ctx.address().downgrade().recipient()))
            .into_actor(self)
            .then(|_, _slf, _| fut::ready(()))
            .spawn(ctx);
    }

    fn stopping(&mut self, _ctx: &mut Self::Context) -> Running {
        println!("🐰 stopping ClientA");
        Running::Stop
    }

    fn stopped(&mut self, _ctx: &mut Self::Context) {
        println!("🐰 stopped ClientA");
    }
}

impl Handler<TimePing> for ClientA {
    type Result = ();

    fn handle(&mut self, msg: TimePing, _ctx: &mut Self::Context) -> Self::Result {
        println!("🐰 ClientA received ping: {:?}", msg.0);
    }
}

#[derive(Debug, Default)]
pub struct ClientB;

impl Actor for ClientB {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        println!("🐇 starting ClientB");
        TimeService::from_registry()
            .send(RegisterForTime(ctx.address().downgrade().recipient()))
            .into_actor(self)
            .then(|_, _slf, _| fut::ready(()))
            .spawn(ctx);
    }

Trait Implementations

impl<A: Actor> Clone for Addr<A>

fn clone(&self) -> Addr<A>

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<A: Actor> Debug for Addr<A>

fn fmt(&self, fmt: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<A, M: Message + Send + 'static> From<Addr<A>> for Recipient<M>

where A: Handler<M> + Actor, M::Result: Send, A::Context: ToEnvelope<A, M>,

fn from(addr: Addr<A>) -> Self

Converts to this type from the input type.

impl<A, M: Message + Send + 'static> From<Addr<A>> for WeakRecipient<M>

where A: Handler<M> + Actor, M::Result: Send, A::Context: ToEnvelope<A, M>,

fn from(addr: Addr<A>) -> WeakRecipient<M>

Converts to this type from the input type.

impl<A: Actor> 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, M, B> MessageResponse<A, M> for Addr<B>

where A: Actor, M: Message<Result = Self>, B: Actor,

fn handle(self, _: &mut A::Context, tx: Option<OneshotSender<Self>>)

impl<A: Actor> PartialEq for Addr<A>

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

This method tests for self and other values to be equal, and is used by ==.

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

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

impl<A: Actor> Eq for Addr<A>