reifydb_runtime/actor/
reply.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4#[cfg(reifydb_single_threaded)]
5use std::cell::RefCell;
6#[cfg(reifydb_single_threaded)]
7use std::rc::Rc;
8
9use cfg_if::cfg_if;
10#[cfg(not(reifydb_single_threaded))]
11use tokio::sync::oneshot;
12
13#[cfg(not(reifydb_single_threaded))]
14use super::mailbox::AskError;
15
16cfg_if! {
17	if #[cfg(reifydb_single_threaded)] {
18
19
20		pub struct Reply<T>(Rc<RefCell<Option<T>>>);
21
22
23		pub struct ReplyReceiver<T>(Rc<RefCell<Option<T>>>);
24
25// SAFETY: DST and WASM are single-threaded, so the Rc/RefCell slot never crosses a thread boundary.
26		unsafe impl<T> Send for Reply<T> {}
27		unsafe impl<T> Sync for Reply<T> {}
28		unsafe impl<T> Send for ReplyReceiver<T> {}
29		unsafe impl<T> Sync for ReplyReceiver<T> {}
30
31
32		pub fn reply_channel<T>() -> (Reply<T>, ReplyReceiver<T>) {
33			let slot = Rc::new(RefCell::new(None));
34			(Reply(Rc::clone(&slot)), ReplyReceiver(slot))
35		}
36
37		impl<T> Reply<T> {
38
39			pub fn send(self, value: T) {
40				*self.0.borrow_mut() = Some(value);
41			}
42		}
43
44		impl<T> ReplyReceiver<T> {
45
46			pub fn try_recv(&self) -> Option<T> {
47				self.0.borrow_mut().take()
48			}
49		}
50	} else {
51
52		pub struct Reply<T>(oneshot::Sender<T>);
53
54
55		pub struct ReplyReceiver<T>(oneshot::Receiver<T>);
56
57
58		pub fn reply_channel<T>() -> (Reply<T>, ReplyReceiver<T>) {
59			let (tx, rx) = oneshot::channel();
60			(Reply(tx), ReplyReceiver(rx))
61		}
62
63		impl<T> Reply<T> {
64
65			pub fn send(self, value: T) {
66				let _ = self.0.send(value);
67			}
68		}
69
70		impl<T> ReplyReceiver<T> {
71
72			pub async fn recv(self) -> Result<T, AskError> {
73				self.0.await.map_err(|_| AskError::ResponseClosed)
74			}
75
76			pub fn blocking_recv(self) -> Result<T, AskError> {
77				self.0.blocking_recv().map_err(|_| AskError::ResponseClosed)
78			}
79		}
80	}
81}
reifydb_runtime/actor/reply.rs

reifydb_runtime/actor/
reply.rs
