lib.rs 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687
  1. #![feature(impl_trait_in_assoc_type)]
  2. use std::{
  3. any::Any,
  4. collections::HashMap,
  5. fmt::Display,
  6. future::{ready, Future, Ready},
  7. marker::PhantomData,
  8. net::IpAddr,
  9. ops::DerefMut,
  10. pin::Pin,
  11. sync::Arc,
  12. };
  13. use btlib::{bterr, crypto::Creds, error::StringError, BlockPath, Result};
  14. use btserde::{field_helpers::smart_ptr, from_slice, to_vec, write_to};
  15. use bttp::{DeserCallback, MsgCallback, Receiver, Replier, Transmitter};
  16. use serde::{de::DeserializeOwned, Deserialize, Serialize};
  17. use tokio::{
  18. sync::{mpsc, oneshot, Mutex, RwLock},
  19. task::JoinHandle,
  20. };
  21. use uuid::Uuid;
  22. /// Declares a new [Runtime] which listens for messages at the given IP address and uses the given
  23. /// [Creds]. Runtimes are intended to be created once in a process's lifetime and continue running
  24. /// until the process exits.
  25. #[macro_export]
  26. macro_rules! declare_runtime {
  27. ($name:ident, $ip_addr:expr, $creds:expr) => {
  28. ::lazy_static::lazy_static! {
  29. static ref $name: &'static ::btrun::Runtime = {
  30. ::lazy_static::lazy_static! {
  31. static ref RUNTIME: ::btrun::Runtime = ::btrun::Runtime::_new($creds).unwrap();
  32. static ref RECEIVER: ::bttp::Receiver = _new_receiver($ip_addr, $creds, &*RUNTIME);
  33. }
  34. // By dereferencing RECEIVER we ensure it is started.
  35. let _ = &*RECEIVER;
  36. &*RUNTIME
  37. };
  38. }
  39. };
  40. }
  41. /// This function is not intended to be called by downstream crates.
  42. #[doc(hidden)]
  43. pub fn _new_receiver<C>(ip_addr: IpAddr, creds: Arc<C>, runtime: &'static Runtime) -> Receiver
  44. where
  45. C: 'static + Send + Sync + Creds,
  46. {
  47. let callback = RuntimeCallback::new(runtime);
  48. Receiver::new(ip_addr, creds, callback).unwrap()
  49. }
  50. /// An actor runtime.
  51. ///
  52. /// Actors can be activated by the runtime and execute autonomously until they return. Running
  53. /// actors can be sent messages using the `send` method, which does not wait for a response from the
  54. /// recipient. If a reply is needed, then `call` can be used, which returns a future that will
  55. /// be ready when the reply has been received.
  56. pub struct Runtime {
  57. path: Arc<BlockPath>,
  58. handles: RwLock<HashMap<Uuid, ActorHandle>>,
  59. peers: RwLock<HashMap<Arc<BlockPath>, Transmitter>>,
  60. }
  61. impl Runtime {
  62. /// This method is not intended to be called directly by downstream crates. Use the macro
  63. /// [declare_runtime] to create a [Runtime].
  64. ///
  65. /// If you create a non-static [Runtime], your process will panic when it is dropped.
  66. #[doc(hidden)]
  67. pub fn _new<C: 'static + Send + Sync + Creds>(creds: Arc<C>) -> Result<Runtime> {
  68. let path = Arc::new(creds.bind_path()?);
  69. Ok(Runtime {
  70. path,
  71. handles: RwLock::new(HashMap::new()),
  72. peers: RwLock::new(HashMap::new()),
  73. })
  74. }
  75. pub fn path(&self) -> &Arc<BlockPath> {
  76. &self.path
  77. }
  78. /// Returns the number of actors that are currently executing in this [Runtime].
  79. pub async fn num_running(&self) -> usize {
  80. let guard = self.handles.read().await;
  81. guard.len()
  82. }
  83. /// Sends a message to the actor identified by the given [ActorName].
  84. pub async fn send<T: 'static + SendMsg>(
  85. &self,
  86. to: ActorName,
  87. from: ActorName,
  88. msg: T,
  89. ) -> Result<()> {
  90. if to.path == self.path {
  91. let guard = self.handles.read().await;
  92. if let Some(handle) = guard.get(&to.act_id) {
  93. handle.send(from, msg).await
  94. } else {
  95. Err(bterr!("invalid actor name"))
  96. }
  97. } else {
  98. let guard = self.peers.read().await;
  99. if let Some(peer) = guard.get(&to.path) {
  100. let buf = to_vec(&msg)?;
  101. let wire_msg = WireMsg {
  102. to,
  103. from,
  104. payload: &buf,
  105. };
  106. peer.send(wire_msg).await
  107. } else {
  108. // TODO: Use the filesystem to discover the address of the recipient and connect to
  109. // it.
  110. todo!()
  111. }
  112. }
  113. }
  114. /// Sends a message to the actor identified by the given [ActorName] and returns a future which
  115. /// is ready when a reply has been received.
  116. pub async fn call<T: 'static + CallMsg>(
  117. &self,
  118. to: ActorName,
  119. from: ActorName,
  120. msg: T,
  121. ) -> Result<T::Reply> {
  122. if to.path == self.path {
  123. let guard = self.handles.read().await;
  124. if let Some(handle) = guard.get(&to.act_id) {
  125. handle.call_through(from, msg).await
  126. } else {
  127. Err(bterr!("invalid actor name"))
  128. }
  129. } else {
  130. let guard = self.peers.read().await;
  131. if let Some(peer) = guard.get(&to.path) {
  132. let buf = to_vec(&msg)?;
  133. let wire_msg = WireMsg {
  134. to,
  135. from,
  136. payload: &buf,
  137. };
  138. peer.call(wire_msg, ReplyCallback::<T>::new()).await?
  139. } else {
  140. // TODO: Use the filesystem to discover the address of the recipient and connect to
  141. // it.
  142. todo!()
  143. }
  144. }
  145. }
  146. /// Resolves the given [ServiceName] to an [ActorName] which is part of it.
  147. pub async fn resolve<'a>(&'a self, _service: &ServiceName) -> Result<ActorName> {
  148. todo!()
  149. }
  150. /// Activates a new actor using the given activator function and returns a handle to it.
  151. pub async fn activate<Msg, F, Fut>(&'static self, activator: F) -> ActorName
  152. where
  153. Msg: 'static + CallMsg,
  154. Fut: 'static + Send + Future<Output = ()>,
  155. F: FnOnce(&'static Runtime, mpsc::Receiver<Envelope<Msg>>, Uuid) -> Fut,
  156. {
  157. let mut guard = self.handles.write().await;
  158. let act_id = {
  159. let mut act_id = Uuid::new_v4();
  160. while guard.contains_key(&act_id) {
  161. act_id = Uuid::new_v4();
  162. }
  163. act_id
  164. };
  165. let act_name = self.actor_name(act_id);
  166. let (tx, rx) = mpsc::channel::<Envelope<Msg>>(MAILBOX_LIMIT);
  167. // The deliverer closure is responsible for deserializing messages received over the wire
  168. // and delivering them to the actor's mailbox, and sending replies to call messages.
  169. let deliverer = {
  170. let buffer = Arc::new(Mutex::new(Vec::<u8>::new()));
  171. let tx = tx.clone();
  172. let act_name = act_name.clone();
  173. move |envelope: WireEnvelope| {
  174. let (wire_msg, replier) = envelope.into_parts();
  175. let result = from_slice(wire_msg.payload);
  176. let buffer = buffer.clone();
  177. let tx = tx.clone();
  178. let act_name = act_name.clone();
  179. let fut: FutureResult = Box::pin(async move {
  180. let msg = result?;
  181. if let Some(mut replier) = replier {
  182. let (envelope, rx) = Envelope::new_call(act_name, msg);
  183. tx.send(envelope).await.map_err(|_| {
  184. bterr!("failed to deliver message. Recipient may have halted.")
  185. })?;
  186. match rx.await {
  187. Ok(reply) => {
  188. let mut guard = buffer.lock().await;
  189. guard.clear();
  190. write_to(&reply, guard.deref_mut())?;
  191. let wire_reply = WireReply::Ok(&guard);
  192. replier.reply(wire_reply).await
  193. }
  194. Err(err) => replier.reply_err(err.to_string(), None).await,
  195. }
  196. } else {
  197. tx.send(Envelope::new_send(act_name, msg))
  198. .await
  199. .map_err(|_| {
  200. bterr!("failed to deliver message. Recipient may have halted.")
  201. })
  202. }
  203. });
  204. fut
  205. }
  206. };
  207. let handle = tokio::task::spawn(activator(self, rx, act_id));
  208. let actor_handle = ActorHandle::new(handle, tx, deliverer);
  209. guard.insert(act_id, actor_handle);
  210. act_name
  211. }
  212. /// Registers an actor as a service with the given [ServiceId].
  213. pub async fn register<Msg, Fut, F, G>(
  214. &self,
  215. _id: ServiceId,
  216. _activator: F,
  217. _deserializer: G,
  218. ) -> Result<()>
  219. where
  220. Msg: 'static + CallMsg,
  221. Fut: 'static + Send + Future<Output = ()>,
  222. F: Fn(mpsc::Receiver<Envelope<Msg>>, Uuid) -> Fut,
  223. G: 'static + Send + Sync + Fn(&[u8]) -> Result<Msg>,
  224. {
  225. todo!()
  226. }
  227. /// Returns the [ActorHandle] for the actor with the given name.
  228. ///
  229. /// If there is no such actor in this runtime then a [RuntimeError::BadActorName] error is
  230. /// returned.
  231. ///
  232. /// Note that the actor will be aborted when the given handle is dropped (unless it has already
  233. /// returned when the handle is dropped), and no further messages will be delivered to it by
  234. /// this runtime.
  235. pub async fn take(&self, name: &ActorName) -> Result<ActorHandle> {
  236. if name.path == self.path {
  237. let mut guard = self.handles.write().await;
  238. if let Some(handle) = guard.remove(&name.act_id) {
  239. Ok(handle)
  240. } else {
  241. Err(RuntimeError::BadActorName(name.clone()).into())
  242. }
  243. } else {
  244. Err(RuntimeError::BadActorName(name.clone()).into())
  245. }
  246. }
  247. /// Returns the name of the actor in this runtime with the given actor ID.
  248. pub fn actor_name(&self, act_id: Uuid) -> ActorName {
  249. ActorName::new(self.path.clone(), act_id)
  250. }
  251. }
  252. impl Drop for Runtime {
  253. fn drop(&mut self) {
  254. panic!("A Runtime was dropped. Panicking to avoid undefined behavior.");
  255. }
  256. }
  257. #[derive(Debug, Clone, PartialEq, Eq)]
  258. pub enum RuntimeError {
  259. BadActorName(ActorName),
  260. }
  261. impl Display for RuntimeError {
  262. fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
  263. match self {
  264. Self::BadActorName(name) => write!(f, "bad actor name: {name}"),
  265. }
  266. }
  267. }
  268. impl std::error::Error for RuntimeError {}
  269. /// Represents the terminal state of an actor, where it stops processing messages and halts.
  270. pub struct End;
  271. impl End {
  272. /// Returns the identifier for this type which is expected in protocol definitions.
  273. pub fn ident() -> &'static str {
  274. stringify!(End)
  275. }
  276. }
  277. #[allow(dead_code)]
  278. /// Delivered to an actor implementation when it starts up.
  279. pub struct Activate {
  280. /// A reference to the `Runtime` which is running this actor.
  281. rt: &'static Runtime,
  282. /// The ID assigned to this actor.
  283. act_id: Uuid,
  284. }
  285. impl Activate {
  286. pub fn new(rt: &'static Runtime, act_id: Uuid) -> Self {
  287. Self { rt, act_id }
  288. }
  289. }
  290. /// Deserializes replies sent over the wire.
  291. pub struct ReplyCallback<T> {
  292. _phantom: PhantomData<T>,
  293. }
  294. impl<T: CallMsg> ReplyCallback<T> {
  295. pub fn new() -> Self {
  296. Self {
  297. _phantom: PhantomData,
  298. }
  299. }
  300. }
  301. impl<T: CallMsg> Default for ReplyCallback<T> {
  302. fn default() -> Self {
  303. Self::new()
  304. }
  305. }
  306. impl<T: CallMsg> DeserCallback for ReplyCallback<T> {
  307. type Arg<'de> = WireReply<'de> where T: 'de;
  308. type Return = Result<T::Reply>;
  309. type CallFut<'de> = Ready<Self::Return> where T: 'de, T::Reply: 'de;
  310. fn call<'de>(&'de mut self, arg: Self::Arg<'de>) -> Self::CallFut<'de> {
  311. let result = match arg {
  312. WireReply::Ok(slice) => from_slice(slice).map_err(|err| err.into()),
  313. WireReply::Err(msg) => Err(StringError::new(msg.to_string()).into()),
  314. };
  315. ready(result)
  316. }
  317. }
  318. struct SendReplyCallback {
  319. replier: Option<Replier>,
  320. }
  321. impl SendReplyCallback {
  322. fn new(replier: Replier) -> Self {
  323. Self {
  324. replier: Some(replier),
  325. }
  326. }
  327. }
  328. impl DeserCallback for SendReplyCallback {
  329. type Arg<'de> = WireReply<'de>;
  330. type Return = Result<()>;
  331. type CallFut<'de> = impl 'de + Future<Output = Self::Return>;
  332. fn call<'de>(&'de mut self, arg: Self::Arg<'de>) -> Self::CallFut<'de> {
  333. async move {
  334. if let Some(mut replier) = self.replier.take() {
  335. replier.reply(arg).await
  336. } else {
  337. Ok(())
  338. }
  339. }
  340. }
  341. }
  342. /// This struct implements the server callback for network messages.
  343. #[derive(Clone)]
  344. struct RuntimeCallback {
  345. rt: &'static Runtime,
  346. }
  347. impl RuntimeCallback {
  348. fn new(rt: &'static Runtime) -> Self {
  349. Self { rt }
  350. }
  351. async fn deliver_local(&self, msg: WireMsg<'_>, replier: Option<Replier>) -> Result<()> {
  352. let guard = self.rt.handles.read().await;
  353. if let Some(handle) = guard.get(&msg.to.act_id) {
  354. let envelope = if let Some(replier) = replier {
  355. WireEnvelope::Call { msg, replier }
  356. } else {
  357. WireEnvelope::Send { msg }
  358. };
  359. (handle.deliverer)(envelope).await
  360. } else {
  361. Err(bterr!("invalid actor name: {}", msg.to))
  362. }
  363. }
  364. async fn route_msg(&self, msg: WireMsg<'_>, replier: Option<Replier>) -> Result<()> {
  365. let guard = self.rt.peers.read().await;
  366. if let Some(tx) = guard.get(msg.to.path()) {
  367. if let Some(replier) = replier {
  368. let callback = SendReplyCallback::new(replier);
  369. tx.call(msg, callback).await?
  370. } else {
  371. tx.send(msg).await
  372. }
  373. } else {
  374. Err(bterr!(
  375. "unable to deliver message to peer at '{}'",
  376. msg.to.path()
  377. ))
  378. }
  379. }
  380. }
  381. impl MsgCallback for RuntimeCallback {
  382. type Arg<'de> = WireMsg<'de>;
  383. type CallFut<'de> = impl 'de + Future<Output = Result<()>>;
  384. fn call<'de>(&'de self, arg: bttp::MsgReceived<Self::Arg<'de>>) -> Self::CallFut<'de> {
  385. async move {
  386. let (_, body, replier) = arg.into_parts();
  387. if body.to.path() == self.rt.path() {
  388. self.deliver_local(body, replier).await
  389. } else {
  390. self.route_msg(body, replier).await
  391. }
  392. }
  393. }
  394. }
  395. /// A unique identifier for a particular service.
  396. #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
  397. pub struct ServiceId(#[serde(with = "smart_ptr")] Arc<String>);
  398. impl From<String> for ServiceId {
  399. fn from(value: String) -> Self {
  400. Self(Arc::new(value))
  401. }
  402. }
  403. impl<'a> From<&'a str> for ServiceId {
  404. fn from(value: &'a str) -> Self {
  405. Self(Arc::new(value.to_owned()))
  406. }
  407. }
  408. /// A unique identifier for a service.
  409. ///
  410. /// A service is a collection of actors in the same directory which provide some functionality.
  411. #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
  412. pub struct ServiceName {
  413. /// The path to the directory containing the service.
  414. #[serde(with = "smart_ptr")]
  415. path: Arc<BlockPath>,
  416. /// The id of the service.
  417. service_id: ServiceId,
  418. }
  419. /// A unique identifier for a specific actor activation.
  420. #[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Serialize, Deserialize)]
  421. pub struct ActorName {
  422. /// The path to the directory containing this actor.
  423. #[serde(with = "smart_ptr")]
  424. path: Arc<BlockPath>,
  425. /// A unique identifier for an actor activation. Even as an actor transitions to different types
  426. /// as it handles messages, this value does not change. Thus this value can be used to trace an
  427. /// actor through a series of state transitions.
  428. act_id: Uuid,
  429. }
  430. impl ActorName {
  431. pub fn new(path: Arc<BlockPath>, act_id: Uuid) -> Self {
  432. Self { path, act_id }
  433. }
  434. pub fn path(&self) -> &Arc<BlockPath> {
  435. &self.path
  436. }
  437. pub fn act_id(&self) -> Uuid {
  438. self.act_id
  439. }
  440. }
  441. impl Display for ActorName {
  442. fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
  443. write!(f, "{}@{}", self.act_id, self.path)
  444. }
  445. }
  446. /// Trait for messages which expect exactly one reply.
  447. pub trait CallMsg: Serialize + DeserializeOwned + Send + Sync {
  448. /// The reply type expected for this message.
  449. type Reply: Serialize + DeserializeOwned + Send + Sync;
  450. }
  451. /// Trait for messages which expect exactly zero replies.
  452. pub trait SendMsg: CallMsg {}
  453. /// A type used to express when a reply is not expected for a message type.
  454. #[derive(Serialize, Deserialize)]
  455. enum NoReply {}
  456. /// The maximum number of messages which can be kept in an actor's mailbox.
  457. const MAILBOX_LIMIT: usize = 32;
  458. /// The type of messages sent over the wire between runtimes.
  459. #[derive(Serialize, Deserialize)]
  460. pub struct WireMsg<'a> {
  461. to: ActorName,
  462. from: ActorName,
  463. payload: &'a [u8],
  464. }
  465. impl<'a> WireMsg<'a> {
  466. pub fn new(to: ActorName, from: ActorName, payload: &'a [u8]) -> Self {
  467. Self { to, from, payload }
  468. }
  469. }
  470. impl<'a> bttp::CallMsg<'a> for WireMsg<'a> {
  471. type Reply<'r> = WireReply<'r>;
  472. }
  473. impl<'a> bttp::SendMsg<'a> for WireMsg<'a> {}
  474. #[derive(Serialize, Deserialize)]
  475. pub enum WireReply<'a> {
  476. Ok(&'a [u8]),
  477. Err(&'a str),
  478. }
  479. /// A wrapper around [WireMsg] which indicates whether a call or send was executed.
  480. enum WireEnvelope<'de> {
  481. Send { msg: WireMsg<'de> },
  482. Call { msg: WireMsg<'de>, replier: Replier },
  483. }
  484. impl<'de> WireEnvelope<'de> {
  485. fn into_parts(self) -> (WireMsg<'de>, Option<Replier>) {
  486. match self {
  487. Self::Send { msg } => (msg, None),
  488. Self::Call { msg, replier } => (msg, Some(replier)),
  489. }
  490. }
  491. }
  492. /// Wrapper around a message type `T` which indicates who the message is from and, if the message
  493. /// was dispatched with `call`, provides a channel to reply to it.
  494. pub struct Envelope<T: CallMsg> {
  495. from: ActorName,
  496. reply: Option<oneshot::Sender<T::Reply>>,
  497. msg: T,
  498. }
  499. impl<T: CallMsg> Envelope<T> {
  500. pub fn new(msg: T, reply: Option<oneshot::Sender<T::Reply>>, from: ActorName) -> Self {
  501. Self { from, reply, msg }
  502. }
  503. /// Creates a new envelope containing the given message which does not expect a reply.
  504. fn new_send(from: ActorName, msg: T) -> Self {
  505. Self {
  506. from,
  507. msg,
  508. reply: None,
  509. }
  510. }
  511. /// Creates a new envelope containing the given message which expects exactly one reply.
  512. fn new_call(from: ActorName, msg: T) -> (Self, oneshot::Receiver<T::Reply>) {
  513. let (tx, rx) = oneshot::channel::<T::Reply>();
  514. let envelope = Self {
  515. from,
  516. msg,
  517. reply: Some(tx),
  518. };
  519. (envelope, rx)
  520. }
  521. /// Returns the name of the actor which sent this message.
  522. pub fn from(&self) -> &ActorName {
  523. &self.from
  524. }
  525. /// Returns a reference to the message in this envelope.
  526. pub fn msg(&self) -> &T {
  527. &self.msg
  528. }
  529. /// Sends a reply to this message.
  530. ///
  531. /// If this message is not expecting a reply, or if this message has already been replied to,
  532. /// then an error is returned.
  533. pub fn reply(&mut self, reply: T::Reply) -> Result<()> {
  534. if let Some(tx) = self.reply.take() {
  535. if tx.send(reply).is_ok() {
  536. Ok(())
  537. } else {
  538. Err(bterr!("failed to send reply"))
  539. }
  540. } else {
  541. Err(bterr!("reply already sent"))
  542. }
  543. }
  544. /// Returns true if this message expects a reply and it has not already been replied to.
  545. pub fn needs_reply(&self) -> bool {
  546. self.reply.is_some()
  547. }
  548. pub fn split(self) -> (T, Option<oneshot::Sender<T::Reply>>, ActorName) {
  549. (self.msg, self.reply, self.from)
  550. }
  551. }
  552. type FutureResult = Pin<Box<dyn Send + Future<Output = Result<()>>>>;
  553. pub struct ActorHandle {
  554. handle: Option<JoinHandle<()>>,
  555. sender: Box<dyn Send + Sync + Any>,
  556. deliverer: Box<dyn Send + Sync + Fn(WireEnvelope<'_>) -> FutureResult>,
  557. }
  558. impl ActorHandle {
  559. fn new<T, F>(handle: JoinHandle<()>, sender: mpsc::Sender<Envelope<T>>, deliverer: F) -> Self
  560. where
  561. T: 'static + CallMsg,
  562. F: 'static + Send + Sync + Fn(WireEnvelope<'_>) -> FutureResult,
  563. {
  564. Self {
  565. handle: Some(handle),
  566. sender: Box::new(sender),
  567. deliverer: Box::new(deliverer),
  568. }
  569. }
  570. fn sender<T: 'static + CallMsg>(&self) -> Result<&mpsc::Sender<Envelope<T>>> {
  571. self.sender
  572. .downcast_ref::<mpsc::Sender<Envelope<T>>>()
  573. .ok_or_else(|| bterr!("unexpected message type"))
  574. }
  575. /// Sends a message to the actor represented by this handle.
  576. pub async fn send<T: 'static + SendMsg>(&self, from: ActorName, msg: T) -> Result<()> {
  577. let sender = self.sender()?;
  578. sender
  579. .send(Envelope::new_send(from, msg))
  580. .await
  581. .map_err(|_| bterr!("failed to enqueue message"))?;
  582. Ok(())
  583. }
  584. pub async fn call_through<T: 'static + CallMsg>(
  585. &self,
  586. from: ActorName,
  587. msg: T,
  588. ) -> Result<T::Reply> {
  589. let sender = self.sender()?;
  590. let (envelope, rx) = Envelope::new_call(from, msg);
  591. sender
  592. .send(envelope)
  593. .await
  594. .map_err(|_| bterr!("failed to enqueue call"))?;
  595. let reply = rx.await?;
  596. Ok(reply)
  597. }
  598. pub async fn returned(&mut self) -> Result<()> {
  599. if let Some(handle) = self.handle.take() {
  600. handle.await?;
  601. }
  602. Ok(())
  603. }
  604. pub fn abort(&mut self) {
  605. if let Some(handle) = self.handle.take() {
  606. handle.abort();
  607. }
  608. }
  609. }
  610. impl Drop for ActorHandle {
  611. fn drop(&mut self) {
  612. self.abort();
  613. }
  614. }