near_async/multithread/
sender.rs

1use crate::messaging::{AsyncSendError, CanSend, CanSendAsync, Handler};
2use crate::multithread::runtime_handle::{MultithreadRuntimeHandle, MultithreadRuntimeMessage};
3use crate::{next_message_sequence_num, pretty_type_name};
4use futures::FutureExt;
5use futures::future::BoxFuture;
6use std::fmt::Debug;
7
8impl<A, M> CanSend<M> for MultithreadRuntimeHandle<A>
9where
10    A: Handler<M> + 'static,
11    M: Debug + Send + 'static,
12{
13    fn send(&self, message: M) {
14        let seq = next_message_sequence_num();
15        let message_type = pretty_type_name::<M>();
16        tracing::trace!(target: "multithread_runtime", seq, message_type, "sending sync message");
17
18        let function = |actor: &mut A| {
19            actor.handle(message);
20        };
21
22        let message = MultithreadRuntimeMessage {
23            seq,
24            enqueued_time_ns: self.instrumentation.current_time(),
25            name: message_type,
26            function: Box::new(function),
27        };
28        if let Err(_) = self.send_message(message) {
29            tracing::info!(target: "multithread_runtime", seq, "ignoring sync message, receiving actor is being shut down");
30        }
31    }
32}
33
34impl<A, M, R> CanSendAsync<M, R> for MultithreadRuntimeHandle<A>
35where
36    A: Handler<M, R> + 'static,
37    M: Debug + Send + 'static,
38    R: Send + 'static,
39{
40    fn send_async(&self, message: M) -> BoxFuture<'static, Result<R, AsyncSendError>> {
41        let seq = next_message_sequence_num();
42        let message_type = pretty_type_name::<M>();
43        tracing::trace!(target: "multithread_runtime", seq, message_type, ?message, "sending async message");
44
45        let (sender, receiver) = tokio::sync::oneshot::channel();
46        let future = async move { receiver.await.map_err(|_| AsyncSendError::Dropped) };
47        let function = move |actor: &mut A| {
48            let result = actor.handle(message);
49            sender.send(result).ok(); // OK if the sender doesn't care about the result anymore.
50        };
51
52        let message = MultithreadRuntimeMessage {
53            seq,
54            enqueued_time_ns: self.instrumentation.current_time(),
55            name: message_type,
56            function: Box::new(function),
57        };
58        if let Err(_) = self.send_message(message) {
59            async { Err(AsyncSendError::Dropped) }.boxed()
60        } else {
61            future.boxed()
62        }
63    }
64}
near_async/multithread/sender.rs

near_async/multithread/
sender.rs
