Call async code from an actor

receive is a plain function, so an actor cannot await. To use async code, for example a database query or an HTTP request, pipe the future to the actor itself with context.pipe_to_self. The future runs on its own Tokio task, and its output arrives as an ordinary message. The actor keeps handling other messages meanwhile.

The examples on this page come from a directory actor which looks up customers in a simulated database.

Pipe a future to the actor

The messages contain the request, Lookup, and the message which carries the result back, Found:

enum Message {
    Lookup {
        id: u64,
        reply_to: ReplyTo<Result<String, UnknownCustomer>>,
    },

    Found {
        customer: Result<String, UnknownCustomer>,
        reply_to: ReplyTo<Result<String, UnknownCustomer>>,
    },

    Checksum {
        data: Vec<u8>,
        reply_to: ReplyTo<Result<u64, JoinError>>,
    },

    Computed {
        checksum: Result<u64, JoinError>,
        reply_to: ReplyTo<Result<u64, JoinError>>,
    },
}

The lookup itself is an ordinary async function:

async fn find_customer(id: u64) -> Result<String, UnknownCustomer> {
    sleep(Duration::from_millis(50)).await;

    match id {
        1 => Ok("Alice".to_string()),
        2 => Ok("Bob".to_string()),
        _ => Err(UnknownCustomer(id)),
    }
}

In receive, the actor pipes the future returned by find_customer to itself:

Message::Lookup { id, reply_to } => {
    context.pipe_to_self(find_customer(id), move |customer| Message::Found {
        customer,
        reply_to,
    });
}

Message::Found { customer, reply_to } => reply_to.reply(customer),

pipe_to_self takes the future and a function which turns its output into a message. Here the function is a closure which also moves the request's ReplyTo into the Found message. When Found arrives, the actor replies. Between Lookup and Found, the actor is free for other messages.

Handle errors

A failure is part of the output. find_customer returns a Result, and the Found message carries it as it is. The actor decides what to do with an error in receive, like with any other message: here it simply passes it on to whoever asked.

Run blocking work

Blocking or CPU-heavy work goes into tokio::task::spawn_blocking. Its JoinHandle is itself a future, so you pipe it the same way:

Message::Checksum { data, reply_to } => {
    let computation = task::spawn_blocking(move || checksum(&data));
    context.pipe_to_self(computation, move |checksum| Message::Computed {
        checksum,
        reply_to,
    });
}

Message::Computed { checksum, reply_to } => reply_to.reply(checksum),

The output is a Result<u64, JoinError>. It is an error if the blocking closure panicked.

What to keep in mind

Other messages come first. The actor handles other messages while the future runs, and the result arrives whenever the future is ready. Write receive so that it does not depend on the result arriving before anything else.

Stopping does not cancel the future. If the actor has terminated by the time the future is ready, the result is dropped and logged as a dead letter.

A restart does not cancel the future either. A restarted actor keeps its mailbox, so the result of a future piped before the restart is received by the new state. If an actor must tell such results apart, it gives each request an identifier and keeps the open ones in its state.

A panic delivers nothing. If the future or the conversion panics, tellus logs the panic at error level, and no message arrives. A ReplyTo captured by the conversion is dropped with it, so an ask waiting for it fails with AskError::NoReply instead of waiting for its timeout.

Run it

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let system = ActorSystem::new(Directory);
    let directory = system.root();

    for id in [1, 2, 3] {
        let customer = directory
            .ask(Duration::from_secs(1), |reply_to| Message::Lookup {
                id,
                reply_to,
            })
            .await
            .context("asking for a customer")?;
        match customer {
            Ok(name) => println!("Customer {id}: {name}"),
            Err(error) => println!("Customer {id}: {error}"),
        }
    }

    let checksum = directory
        .ask(Duration::from_secs(1), |reply_to| Message::Checksum {
            data: b"tellus".to_vec(),
            reply_to,
        })
        .await
        .context("asking for a checksum")?
        .context("computing the checksum")?;
    println!("Checksum: {checksum:x}");

    system.stop();
    system
        .terminated()
        .await
        .context("awaiting actor system termination")
}

From a clone of the tellus repository:

cargo run --quiet -p tellus --example how_to_call_async_code

Customer 1: Alice
Customer 2: Bob
Customer 3: unknown customer 3
Checksum: 9c3434cbfbedb0d2