runtime_tests.rs 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755
  1. #![feature(impl_trait_in_assoc_type)]
  2. use btrun::model::*;
  3. use btrun::test_setup;
  4. use btrun::*;
  5. use btlib::Result;
  6. use btproto::protocol;
  7. use log;
  8. use once_cell::sync::Lazy;
  9. use serde::{Deserialize, Serialize};
  10. use std::{
  11. future::{ready, Future, Ready},
  12. sync::{
  13. atomic::{AtomicU8, Ordering},
  14. Arc,
  15. },
  16. };
  17. test_setup!();
  18. mod ping_pong {
  19. use super::*;
  20. use btlib::bterr;
  21. // The following code is a proof-of-concept for what types should be generated for a
  22. // simple ping-pong protocol:
  23. protocol! {
  24. named PingProtocol;
  25. let server = [Server];
  26. let client = [Client];
  27. Client -> End, >service(Server)!Ping;
  28. Server?Ping -> End, >Client!Ping::Reply;
  29. }
  30. //
  31. // In words, the protocol is described as follows.
  32. // 1. When the Listening state receives the Ping message it returns the End state and a
  33. // Ping::Reply message to be sent to the SentPing state.
  34. // 2. When the SentPing state receives the Ping::Reply message it returns the End state.
  35. //
  36. // The End state represents an end to the session described by the protocol. When an actor
  37. // transitions to the End state its function returns.
  38. // When a state is expecting a Reply message, an error occurs if the message is not received
  39. // in a timely manner.
  40. enum PingClientState<T: Client> {
  41. Client(T),
  42. End(End),
  43. }
  44. impl<T: Client> PingClientState<T> {
  45. const fn name(&self) -> &'static str {
  46. match self {
  47. Self::Client(_) => "Client",
  48. Self::End(_) => "End",
  49. }
  50. }
  51. }
  52. struct ClientHandle<T: Client> {
  53. state: Option<PingClientState<T>>,
  54. runtime: &'static Runtime,
  55. }
  56. impl<T: Client> ClientHandle<T> {
  57. async fn send_ping(&mut self, mut msg: Ping, service: ServiceAddr) -> Result<PingReply> {
  58. let state = self
  59. .state
  60. .take()
  61. .ok_or_else(|| bterr!("State was not returned."))?;
  62. let (new_state, result) = match state {
  63. PingClientState::Client(state) => match state.on_send_ping(&mut msg).await {
  64. TransResult::Ok((new_state, _)) => {
  65. let new_state = PingClientState::End(new_state);
  66. let result = self
  67. .runtime
  68. .call_service(service, PingProtocolMsgs::Ping(msg))
  69. .await;
  70. (new_state, result)
  71. }
  72. TransResult::Abort { from, err } => {
  73. let new_state = PingClientState::Client(from);
  74. (new_state, Err(err))
  75. }
  76. TransResult::Fatal { err } => return Err(err),
  77. },
  78. state => {
  79. let result = Err(bterr!("Can't send Ping in state {}.", state.name()));
  80. (state, result)
  81. }
  82. };
  83. self.state = Some(new_state);
  84. let reply = result?;
  85. match reply {
  86. PingProtocolMsgs::PingReply(reply) => Ok(reply),
  87. msg => Err(bterr!(
  88. "Unexpected message type sent in reply: {}",
  89. msg.name()
  90. )),
  91. }
  92. }
  93. }
  94. async fn spawn_client<T: Client>(init: T, runtime: &'static Runtime) -> ClientHandle<T> {
  95. let state = Some(PingClientState::Client(init));
  96. ClientHandle { state, runtime }
  97. }
  98. async fn register_server<Init, F>(
  99. make_init: F,
  100. rt: &'static Runtime,
  101. id: ServiceId,
  102. ) -> Result<ServiceName>
  103. where
  104. Init: 'static + Server,
  105. F: 'static + Send + Sync + Clone + Fn() -> Init,
  106. {
  107. enum ServerState<S> {
  108. Server(S),
  109. End(End),
  110. }
  111. impl<S> Named for ServerState<S> {
  112. fn name(&self) -> Arc<String> {
  113. static SERVER_NAME: Lazy<Arc<String>> = Lazy::new(|| Arc::new("Server".into()));
  114. static END_NAME: Lazy<Arc<String>> = Lazy::new(|| Arc::new("End".into()));
  115. match self {
  116. Self::Server(_) => SERVER_NAME.clone(),
  117. Self::End(_) => END_NAME.clone(),
  118. }
  119. }
  120. }
  121. async fn server_loop<Init, F>(
  122. _runtime: &'static Runtime,
  123. make_init: F,
  124. mut mailbox: Mailbox<PingProtocolMsgs>,
  125. actor_id: ActorId,
  126. ) -> ActorResult
  127. where
  128. Init: 'static + Server,
  129. F: 'static + Send + Sync + FnOnce() -> Init,
  130. {
  131. let mut state = ServerState::Server(make_init());
  132. while let Some(envelope) = mailbox.recv().await {
  133. state = match envelope {
  134. Envelope::Call {
  135. msg,
  136. reply: replier,
  137. ..
  138. } => match (state, msg) {
  139. (ServerState::Server(listening_state), PingProtocolMsgs::Ping(msg)) => {
  140. match listening_state.handle_ping(msg).await {
  141. TransResult::Ok((new_state, reply)) => {
  142. let replier =
  143. replier.expect("The reply has already been sent.");
  144. if let Err(_) = replier.send(PingProtocolMsgs::PingReply(reply))
  145. {
  146. return Err(ActorError::new(
  147. bterr!("Failed to send Ping reply."),
  148. ActorErrorCommon {
  149. actor_id,
  150. actor_impl: Init::actor_impl(),
  151. state: Init::state_name(),
  152. message: PingProtocolMsgKinds::Ping.name(),
  153. kind: TransKind::Receive,
  154. },
  155. ));
  156. }
  157. ServerState::End(new_state)
  158. }
  159. TransResult::Abort { from, err } => {
  160. log::warn!("Aborted transition from the {} while handling the {} message: {}", "Server", "Ping", err);
  161. ServerState::Server(from)
  162. }
  163. TransResult::Fatal { err } => {
  164. return Err(ActorError::new(
  165. err,
  166. ActorErrorCommon {
  167. actor_id,
  168. actor_impl: Init::actor_impl(),
  169. state: Init::state_name(),
  170. message: PingProtocolMsgKinds::Ping.name(),
  171. kind: TransKind::Receive,
  172. },
  173. ));
  174. }
  175. }
  176. }
  177. (state, _) => state,
  178. },
  179. envelope => {
  180. return Err(ActorError::new(
  181. bterr!("Unexpected envelope type: {}", envelope.name()),
  182. ActorErrorCommon {
  183. actor_id,
  184. actor_impl: Init::actor_impl(),
  185. state: state.name(),
  186. message: envelope.msg_name(),
  187. kind: TransKind::Receive,
  188. },
  189. ))
  190. }
  191. };
  192. if let ServerState::End(_) = state {
  193. break;
  194. }
  195. }
  196. Ok(actor_id)
  197. }
  198. rt.register::<PingProtocolMsgs, _>(id, move |runtime| {
  199. let make_init = make_init.clone();
  200. let fut = async move {
  201. let actor_impl = runtime
  202. .spawn(None, move |mailbox, act_id, runtime| {
  203. server_loop(runtime, make_init, mailbox, act_id)
  204. })
  205. .await;
  206. Ok(actor_impl)
  207. };
  208. Box::pin(fut)
  209. })
  210. .await
  211. }
  212. #[derive(Serialize, Deserialize)]
  213. pub struct Ping;
  214. impl CallMsg for Ping {
  215. type Reply = PingReply;
  216. }
  217. #[derive(Serialize, Deserialize)]
  218. pub struct PingReply;
  219. struct ClientState {
  220. counter: Arc<AtomicU8>,
  221. }
  222. impl ClientState {
  223. fn new(counter: Arc<AtomicU8>) -> Self {
  224. counter.fetch_add(1, Ordering::SeqCst);
  225. Self { counter }
  226. }
  227. }
  228. impl Client for ClientState {
  229. actor_name!("ping_client");
  230. type OnSendPingFut = impl Future<Output = TransResult<Self, (End, PingReply)>>;
  231. fn on_send_ping(self, _msg: &mut Ping) -> Self::OnSendPingFut {
  232. self.counter.fetch_sub(1, Ordering::SeqCst);
  233. ready(TransResult::Ok((End, PingReply)))
  234. }
  235. }
  236. struct ServerState {
  237. counter: Arc<AtomicU8>,
  238. }
  239. impl ServerState {
  240. fn new(counter: Arc<AtomicU8>) -> Self {
  241. counter.fetch_add(1, Ordering::SeqCst);
  242. Self { counter }
  243. }
  244. }
  245. impl Server for ServerState {
  246. actor_name!("ping_server");
  247. type HandlePingFut = impl Future<Output = TransResult<Self, (End, PingReply)>>;
  248. fn handle_ping(self, _msg: Ping) -> Self::HandlePingFut {
  249. self.counter.fetch_sub(1, Ordering::SeqCst);
  250. ready(TransResult::Ok((End, PingReply)))
  251. }
  252. }
  253. #[test]
  254. fn ping_pong_test() {
  255. ASYNC_RT.block_on(async {
  256. const SERVICE_ID: &str = "PingPongProtocolServer";
  257. let service_id = ServiceId::from(SERVICE_ID);
  258. let counter = Arc::new(AtomicU8::new(0));
  259. let service_name = {
  260. let service_counter = counter.clone();
  261. let make_init = move || {
  262. let server_counter = service_counter.clone();
  263. ServerState::new(server_counter)
  264. };
  265. register_server(make_init, &RUNTIME, service_id.clone())
  266. .await
  267. .unwrap()
  268. };
  269. let mut client_handle = spawn_client(ClientState::new(counter.clone()), &RUNTIME).await;
  270. let service_addr = ServiceAddr::new(service_name, true);
  271. client_handle.send_ping(Ping, service_addr).await.unwrap();
  272. assert_eq!(0, counter.load(Ordering::SeqCst));
  273. RUNTIME.deregister(&service_id, None).await.unwrap();
  274. });
  275. }
  276. }
  277. mod travel_agency {
  278. use super::*;
  279. // Here's another protocol example. This is the Customer and Travel Agency protocol used as an
  280. // example in the survey paper "Behavioral Types in Programming Languages."
  281. // Note that the Choosing state can send messages at any time, not just in response to another
  282. // message because there is a transition from Choosing that doesn't use the receive operator
  283. // (`?`).
  284. protocol! {
  285. named TravelAgency;
  286. let agency = [Listening];
  287. let customer = [Choosing];
  288. Choosing -> Choosing, >service(Listening)!Query;
  289. Choosing -> Choosing, >service(Listening)!Accept;
  290. Choosing -> Choosing, >service(Listening)!Reject;
  291. Listening?Query -> Listening, >Choosing!Query::Reply;
  292. Choosing?Query::Reply -> Choosing;
  293. Listening?Accept -> End, >Choosing!Accept::Reply;
  294. Choosing?Accept::Reply -> End;
  295. Listening?Reject -> End, >Choosing!Reject::Reply;
  296. Choosing?Reject::Reply -> End;
  297. }
  298. #[derive(Serialize, Deserialize)]
  299. pub struct Query;
  300. impl CallMsg for Query {
  301. type Reply = ();
  302. }
  303. #[derive(Serialize, Deserialize)]
  304. pub struct Reject;
  305. impl CallMsg for Reject {
  306. type Reply = ();
  307. }
  308. #[derive(Serialize, Deserialize)]
  309. pub struct Accept;
  310. impl CallMsg for Accept {
  311. type Reply = ();
  312. }
  313. }
  314. #[allow(dead_code)]
  315. mod client_callback {
  316. use super::*;
  317. use btlib::bterr;
  318. use once_cell::sync::Lazy;
  319. use std::{panic::panic_any, time::Duration};
  320. use tokio::{sync::oneshot, time::timeout};
  321. #[derive(Serialize, Deserialize)]
  322. pub struct Register {
  323. factor: usize,
  324. }
  325. #[derive(Serialize, Deserialize)]
  326. pub struct Completed {
  327. value: usize,
  328. }
  329. protocol! {
  330. named ClientCallback;
  331. let server = [Listening];
  332. let worker = [Working];
  333. let client = [Unregistered, Registered];
  334. Unregistered -> Registered, >service(Listening)!Register[Registered];
  335. Listening?Register[Registered] -> Listening, Working[Registered];
  336. Working[Registered] -> End, >Registered!Completed;
  337. Registered?Completed -> End;
  338. }
  339. struct UnregisteredState {
  340. sender: oneshot::Sender<usize>,
  341. }
  342. impl Unregistered for UnregisteredState {
  343. actor_name!("callback_client");
  344. type OnSendRegisterRegistered = RegisteredState;
  345. type OnSendRegisterFut = Ready<TransResult<Self, Self::OnSendRegisterRegistered>>;
  346. fn on_send_register(self, _arg: &mut Register) -> Self::OnSendRegisterFut {
  347. ready(TransResult::Ok(RegisteredState {
  348. sender: self.sender,
  349. }))
  350. }
  351. }
  352. struct RegisteredState {
  353. sender: oneshot::Sender<usize>,
  354. }
  355. impl Registered for RegisteredState {
  356. type HandleCompletedFut = Ready<TransResult<Self, End>>;
  357. fn handle_completed(self, arg: Completed) -> Self::HandleCompletedFut {
  358. self.sender.send(arg.value).unwrap();
  359. ready(TransResult::Ok(End))
  360. }
  361. }
  362. struct ListeningState {
  363. multiple: usize,
  364. }
  365. impl Listening for ListeningState {
  366. actor_name!("callback_server");
  367. type HandleRegisterListening = ListeningState;
  368. type HandleRegisterWorking = WorkingState;
  369. type HandleRegisterFut = Ready<TransResult<Self, (ListeningState, WorkingState)>>;
  370. fn handle_register(self, arg: Register) -> Self::HandleRegisterFut {
  371. let multiple = self.multiple;
  372. ready(TransResult::Ok((
  373. self,
  374. WorkingState {
  375. factor: arg.factor,
  376. multiple,
  377. },
  378. )))
  379. }
  380. }
  381. struct WorkingState {
  382. factor: usize,
  383. multiple: usize,
  384. }
  385. impl Working for WorkingState {
  386. actor_name!("callback_worker");
  387. type OnSendCompletedFut = Ready<TransResult<Self, (End, Completed)>>;
  388. fn on_send_completed(self) -> Self::OnSendCompletedFut {
  389. let value = self.multiple * self.factor;
  390. ready(TransResult::Ok((End, Completed { value })))
  391. }
  392. }
  393. use ::tokio::sync::Mutex;
  394. enum ClientState<Init: Unregistered> {
  395. Unregistered(Init),
  396. Registered(Init::OnSendRegisterRegistered),
  397. End(End),
  398. }
  399. impl<Init: Unregistered> Named for ClientState<Init> {
  400. fn name(&self) -> Arc<String> {
  401. static UNREGISTERED_NAME: Lazy<Arc<String>> =
  402. Lazy::new(|| Arc::new("Unregistered".into()));
  403. static REGISTERED_NAME: Lazy<Arc<String>> = Lazy::new(|| Arc::new("Registered".into()));
  404. static END_NAME: Lazy<Arc<String>> = Lazy::new(|| Arc::new("End".into()));
  405. match self {
  406. Self::Unregistered(_) => UNREGISTERED_NAME.clone(),
  407. Self::Registered(_) => REGISTERED_NAME.clone(),
  408. Self::End(_) => END_NAME.clone(),
  409. }
  410. }
  411. }
  412. struct ClientHandle<Init: Unregistered> {
  413. runtime: &'static Runtime,
  414. state: Arc<Mutex<Option<ClientState<Init>>>>,
  415. name: ActorName,
  416. }
  417. impl<Init: Unregistered> ClientHandle<Init> {
  418. async fn send_register(&self, to: ServiceAddr, mut msg: Register) -> Result<()> {
  419. let mut guard = self.state.lock().await;
  420. let state = guard
  421. .take()
  422. .unwrap_or_else(|| panic!("Logic error. The state was not returned."));
  423. let new_state = match state {
  424. ClientState::Unregistered(state) => match state.on_send_register(&mut msg).await {
  425. TransResult::Ok(new_state) => {
  426. let msg = ClientCallbackMsgs::Register(msg);
  427. self.runtime
  428. .send_service(to, self.name.clone(), msg)
  429. .await?;
  430. ClientState::Registered(new_state)
  431. }
  432. TransResult::Abort { from, err } => {
  433. log::warn!(
  434. "Aborted transition from the {} state: {}",
  435. "Unregistered",
  436. err
  437. );
  438. ClientState::Unregistered(from)
  439. }
  440. TransResult::Fatal { err } => {
  441. return Err(err);
  442. }
  443. },
  444. state => state,
  445. };
  446. *guard = Some(new_state);
  447. Ok(())
  448. }
  449. }
  450. async fn spawn_client<Init>(init: Init, runtime: &'static Runtime) -> ClientHandle<Init>
  451. where
  452. Init: 'static + Unregistered,
  453. {
  454. let state = Arc::new(Mutex::new(Some(ClientState::Unregistered(init))));
  455. let name = {
  456. let state = state.clone();
  457. runtime.spawn(None, move |mut mailbox, actor_id, _| async move {
  458. while let Some(envelope) = mailbox.recv().await {
  459. let mut guard = state.lock().await;
  460. let state = guard.take()
  461. .unwrap_or_else(|| panic!("Logic error. The state was not returned."));
  462. let new_state = match envelope {
  463. Envelope::Send { msg, .. } => {
  464. match (state, msg) {
  465. (ClientState::Registered(curr_state), ClientCallbackMsgs::Completed(msg)) => {
  466. match curr_state.handle_completed(msg).await {
  467. TransResult::Ok(next) => ClientState::<Init>::End(next),
  468. TransResult::Abort { from, err } => {
  469. log::warn!("Aborted transition from the {} state while handling the {} message: {}", "Registered", "Completed", err);
  470. ClientState::Registered(from)
  471. }
  472. TransResult::Fatal { err } => {
  473. panic_any(ActorError::new(
  474. err,
  475. ActorErrorCommon {
  476. actor_id,
  477. actor_impl: Init::actor_impl(),
  478. state: Init::OnSendRegisterRegistered::state_name(),
  479. message: ClientCallbackMsgKinds::Completed.name(),
  480. kind: TransKind::Receive,
  481. }));
  482. }
  483. }
  484. }
  485. (state, msg) => {
  486. log::error!("Unexpected message {} in state {}.", msg.name(), state.name());
  487. state
  488. }
  489. }
  490. }
  491. envelope => return Err(ActorError::new(
  492. bterr!("Unexpected envelope type: {}", envelope.name()),
  493. ActorErrorCommon {
  494. actor_id,
  495. actor_impl: Init::actor_impl(),
  496. state: state.name(),
  497. message: envelope.msg_name(),
  498. kind: TransKind::Receive,
  499. }))
  500. };
  501. *guard = Some(new_state);
  502. }
  503. Ok(actor_id)
  504. }).await
  505. };
  506. ClientHandle {
  507. runtime,
  508. state,
  509. name,
  510. }
  511. }
  512. async fn register_server<Init, F>(
  513. make_init: F,
  514. runtime: &'static Runtime,
  515. service_id: ServiceId,
  516. ) -> Result<ServiceName>
  517. where
  518. Init: 'static + Listening<HandleRegisterListening = Init>,
  519. F: 'static + Send + Sync + Clone + Fn() -> Init,
  520. {
  521. enum ServerState<S: Listening> {
  522. Listening(S),
  523. }
  524. impl<S: Listening> Named for ServerState<S> {
  525. fn name(&self) -> Arc<String> {
  526. static LISTENING_NAME: Lazy<Arc<String>> =
  527. Lazy::new(|| Arc::new("Listening".into()));
  528. match self {
  529. Self::Listening(_) => LISTENING_NAME.clone(),
  530. }
  531. }
  532. }
  533. async fn server_loop<Init, F>(
  534. runtime: &'static Runtime,
  535. make_init: F,
  536. mut mailbox: Mailbox<ClientCallbackMsgs>,
  537. actor_id: ActorId,
  538. ) -> ActorResult
  539. where
  540. Init: 'static + Listening<HandleRegisterListening = Init>,
  541. F: 'static + Send + Sync + Fn() -> Init,
  542. {
  543. let mut state = ServerState::Listening(make_init());
  544. while let Some(envelope) = mailbox.recv().await {
  545. let new_state = match envelope {
  546. Envelope::Send { msg, from, .. } => match (state, msg) {
  547. (ServerState::Listening(curr_state), ClientCallbackMsgs::Register(msg)) => {
  548. match curr_state.handle_register(msg).await {
  549. TransResult::Ok((new_state, working_state)) => {
  550. start_worker(working_state, from, runtime).await;
  551. ServerState::Listening(new_state)
  552. }
  553. TransResult::Abort { from, err } => {
  554. log::warn!("Aborted transition from the {} state while handling the {} message: {}", "Listening", "Register", err);
  555. ServerState::Listening(from)
  556. }
  557. TransResult::Fatal { err } => {
  558. let err = ActorError::new(
  559. err,
  560. ActorErrorCommon {
  561. actor_id,
  562. actor_impl: Init::actor_impl(),
  563. state: Init::state_name(),
  564. message: ClientCallbackMsgKinds::Register.name(),
  565. kind: TransKind::Receive,
  566. },
  567. );
  568. panic_any(format!("{err}"));
  569. }
  570. }
  571. }
  572. (state, msg) => {
  573. log::error!(
  574. "Unexpected message {} in state {}.",
  575. msg.name(),
  576. state.name()
  577. );
  578. state
  579. }
  580. },
  581. envelope => {
  582. return Err(ActorError::new(
  583. bterr!("Unexpected envelope type: {}", envelope.name()),
  584. ActorErrorCommon {
  585. actor_id,
  586. actor_impl: Init::actor_impl(),
  587. state: state.name(),
  588. message: envelope.msg_name(),
  589. kind: TransKind::Receive,
  590. },
  591. ))
  592. }
  593. };
  594. state = new_state;
  595. }
  596. Ok(actor_id)
  597. }
  598. runtime
  599. .register::<ClientCallbackMsgs, _>(service_id, move |runtime: &'static Runtime| {
  600. let make_init = make_init.clone();
  601. let fut = async move {
  602. let make_init = make_init.clone();
  603. let actor_impl = runtime
  604. .spawn(None, move |mailbox, act_id, runtime| {
  605. server_loop(runtime, make_init, mailbox, act_id)
  606. })
  607. .await;
  608. Ok(actor_impl)
  609. };
  610. Box::pin(fut)
  611. })
  612. .await
  613. }
  614. async fn start_worker<Init>(
  615. init: Init,
  616. owned: ActorName,
  617. runtime: &'static Runtime,
  618. ) -> ActorName
  619. where
  620. Init: 'static + Working,
  621. {
  622. enum WorkerState<S: Working> {
  623. Working(S),
  624. }
  625. runtime
  626. .spawn::<ClientCallbackMsgs, _, _>(None, move |_, actor_id, _| async move {
  627. let msg = match init.on_send_completed().await {
  628. TransResult::Ok((End, msg)) => msg,
  629. TransResult::Abort { err, .. } | TransResult::Fatal { err } => {
  630. let err = ActorError::new(
  631. err,
  632. ActorErrorCommon {
  633. actor_id,
  634. actor_impl: Init::actor_impl(),
  635. state: Init::state_name(),
  636. message: ClientCallbackMsgKinds::Completed.name(),
  637. kind: TransKind::Send,
  638. },
  639. );
  640. panic_any(format!("{err}"))
  641. }
  642. };
  643. let from = runtime.actor_name(actor_id);
  644. let msg = ClientCallbackMsgs::Completed(msg);
  645. runtime.send(owned, from, msg).await.unwrap_or_else(|err| {
  646. let err = ActorError::new(
  647. err,
  648. ActorErrorCommon {
  649. actor_id,
  650. actor_impl: Init::actor_impl(),
  651. state: Init::state_name(),
  652. message: ClientCallbackMsgKinds::Completed.name(),
  653. kind: TransKind::Send,
  654. },
  655. );
  656. panic_any(format!("{err}"));
  657. });
  658. Ok(actor_id)
  659. })
  660. .await
  661. }
  662. #[test]
  663. fn client_callback_protocol() {
  664. ASYNC_RT.block_on(async {
  665. const SERVICE_ID: &str = "ClientCallbackProtocolListening";
  666. let service_id = ServiceId::from(SERVICE_ID);
  667. let service_name = {
  668. let make_init = move || ListeningState { multiple: 2 };
  669. register_server(make_init, &RUNTIME, service_id.clone())
  670. .await
  671. .unwrap()
  672. };
  673. let (sender, receiver) = oneshot::channel();
  674. let client_handle = spawn_client(UnregisteredState { sender }, &RUNTIME).await;
  675. let service_addr = ServiceAddr::new(service_name, false);
  676. client_handle
  677. .send_register(service_addr, Register { factor: 21 })
  678. .await
  679. .unwrap();
  680. let value = timeout(Duration::from_millis(500), receiver)
  681. .await
  682. .unwrap()
  683. .unwrap();
  684. assert_eq!(42, value);
  685. });
  686. }
  687. }