compio_actor\mailbox/
call.rs

1use std::fmt;
2
3use futures_channel::oneshot;
4
5use super::{Broker, CallError, DeliverError, Mailbox};
6use crate::{Actor, Handler, Message};
7
8impl<A: Actor> Mailbox<A> {
9    /// Sends a request and waits for the actor's reply.
10    pub async fn call<M, R>(&self, message: M) -> Result<R, CallError<M>>
11    where
12        A: Handler<Call<M, R>>,
13        M: Message,
14        R: Message,
15    {
16        call_with(message, |call| self.inner.send(call)).await
17    }
18}
19
20impl<M: Message, R: Message> Broker<Call<M, R>> {
21    /// Sends a request and waits for the actor's reply.
22    pub async fn call(&self, message: M) -> Result<R, CallError<M>> {
23        call_with(message, |call| self.send(call)).await
24    }
25}
26
27/// A request together with the channel used to answer it.
28pub struct Call<M: Message, R: Message> {
29    message: M,
30    reply: Reply<R>,
31}
32
33impl<M: Message, R: Message> Call<M, R> {
34    fn new(message: M, sender: oneshot::Sender<R>) -> Self {
35        Self {
36            message,
37            reply: Reply(sender),
38        }
39    }
40
41    /// Get the request message.
42    pub fn message(&self) -> &M {
43        &self.message
44    }
45
46    /// Answers the call, returning the response if the caller stopped waiting.
47    pub fn reply(self, response: R) -> Result<(), R> {
48        self.reply.reply(response)
49    }
50
51    /// Splits the owned request from its reply capability.
52    pub fn into_parts(self) -> (M, Reply<R>) {
53        (self.message, self.reply)
54    }
55
56    pub(super) fn into_message(self) -> M {
57        self.message
58    }
59}
60
61impl<M: Message + fmt::Debug, R: Message> fmt::Debug for Call<M, R> {
62    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
63        f.debug_struct("Call")
64            .field("message", &self.message)
65            .finish_non_exhaustive()
66    }
67}
68
69/// A port used to send a reply to a [`Call`].
70pub struct Reply<R: Message>(oneshot::Sender<R>);
71
72impl<R: Message> Reply<R> {
73    /// Answers the call, returning the response if the caller stopped waiting.
74    pub fn reply(self, response: R) -> Result<(), R> {
75        self.0.send(response)
76    }
77}
78
79impl<R: Message> fmt::Debug for Reply<R> {
80    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
81        f.debug_struct("Reply").finish_non_exhaustive()
82    }
83}
84
85pub(crate) async fn call_with<M, R>(
86    message: M,
87    send: impl FnOnce(Call<M, R>) -> Result<(), DeliverError<Call<M, R>>>,
88) -> Result<R, CallError<M>>
89where
90    M: Message,
91    R: Message,
92{
93    let (sender, receiver) = oneshot::channel();
94    send(Call::new(message, sender)).map_err(CallError::from_deliver)?;
95    receiver.await.map_err(|_| CallError::NoReply)
96}
compio_actor\mailbox/call.rs

compio_actor\mailbox/
call.rs
