reifydb_runtime/actor/mailbox/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4#[cfg(all(reifydb_single_threaded, not(reifydb_dst)))]
5use std::cell::{Cell, RefCell};
6use std::fmt;
7#[cfg(all(reifydb_single_threaded, not(reifydb_dst)))]
8use std::rc::Rc;
9#[cfg(all(reifydb_single_threaded, not(reifydb_dst)))]
10use std::sync::Arc;
11#[cfg(all(reifydb_single_threaded, not(reifydb_dst)))]
12use std::sync::atomic::AtomicBool;
13
14use cfg_if::cfg_if;
15
16#[cfg(not(reifydb_single_threaded))]
17pub(crate) mod host;
18
19#[cfg(all(reifydb_single_threaded, not(reifydb_dst)))]
20pub(crate) mod wasm;
21
22#[cfg(reifydb_dst)]
23pub(crate) mod dst;
24
25cfg_if! {
26	if #[cfg(reifydb_dst)] {
27		type ActorRefInnerImpl<M> = dst::ActorRefInner<M>;
28	} else if #[cfg(not(reifydb_single_threaded))] {
29		type ActorRefInnerImpl<M> = host::ActorRefInner<M>;
30	} else {
31		type ActorRefInnerImpl<M> = wasm::ActorRefInner<M>;
32	}
33}
34
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub enum SendError<M> {
37	Closed(M),
38
39	Full(M),
40}
41
42impl<M> SendError<M> {
43	#[inline]
44	pub fn into_inner(self) -> M {
45		match self {
46			SendError::Closed(m) => m,
47			SendError::Full(m) => m,
48		}
49	}
50}
51
52impl<M: fmt::Debug> fmt::Display for SendError<M> {
53	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
54		match self {
55			SendError::Closed(_) => write!(f, "actor mailbox closed"),
56			SendError::Full(_) => write!(f, "actor mailbox full"),
57		}
58	}
59}
60
61impl<M: fmt::Debug> error::Error for SendError<M> {}
62
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub enum AskError {
65	SendFailed,
66
67	ResponseClosed,
68}
69
70impl fmt::Display for AskError {
71	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
72		match self {
73			AskError::SendFailed => write!(f, "failed to send ask request"),
74			AskError::ResponseClosed => write!(f, "response channel closed"),
75		}
76	}
77}
78
79impl error::Error for AskError {}
80
81pub struct ActorRef<M> {
82	inner: ActorRefInnerImpl<M>,
83}
84
85impl<M> Clone for ActorRef<M> {
86	#[inline]
87	fn clone(&self) -> Self {
88		Self {
89			inner: self.inner.clone(),
90		}
91	}
92}
93
94impl<M> fmt::Debug for ActorRef<M> {
95	#[inline]
96	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
97		self.inner.fmt(f)
98	}
99}
100
101// SAFETY: under reifydb_single_threaded there is no second thread, so an ActorRef can never be reached
102// from another one. These impls only satisfy Send + Sync bounds and never back a real transfer.
103#[cfg(reifydb_single_threaded)]
104unsafe impl<M> Send for ActorRef<M> {}
105
106#[cfg(reifydb_single_threaded)]
107unsafe impl<M> Sync for ActorRef<M> {}
108
109impl<M> ActorRef<M> {
110	#[inline]
111	pub(crate) fn from_inner(inner: ActorRefInnerImpl<M>) -> Self {
112		Self {
113			inner,
114		}
115	}
116}
117
118#[cfg(not(reifydb_single_threaded))]
119impl<M: Send> ActorRef<M> {
120	#[inline]
121	pub(crate) fn new(tx: Sender<M>) -> Self {
122		Self {
123			inner: host::ActorRefInner::new(tx),
124		}
125	}
126
127	#[inline]
128	pub(crate) fn set_notify(&self, f: sync::Arc<dyn Fn() + Send + Sync>) {
129		self.inner.set_notify(f)
130	}
131
132	#[inline]
133	pub fn send(&self, msg: M) -> Result<(), SendError<M>> {
134		self.inner.send(msg)
135	}
136
137	#[inline]
138	pub fn send_blocking(&self, msg: M) -> Result<(), SendError<M>> {
139		self.inner.send_blocking(msg)
140	}
141
142	#[inline]
143	pub fn is_alive(&self) -> bool {
144		self.inner.is_alive()
145	}
146}
147
148#[cfg(reifydb_dst)]
149impl<M> ActorRef<M> {
150	#[inline]
151	pub fn send(&self, msg: M) -> Result<(), SendError<M>> {
152		self.inner.send(msg)
153	}
154
155	#[inline]
156	pub fn send_blocking(&self, msg: M) -> Result<(), SendError<M>> {
157		self.inner.send_blocking(msg)
158	}
159
160	#[inline]
161	pub fn is_alive(&self) -> bool {
162		self.inner.is_alive()
163	}
164
165	#[inline]
166	pub(crate) fn mark_stopped(&self) {
167		self.inner.mark_stopped()
168	}
169
170	#[inline]
171	pub(crate) fn set_notify(&self, f: Box<dyn Fn()>) {
172		self.inner.set_notify(f)
173	}
174}
175
176#[cfg(all(reifydb_single_threaded, not(reifydb_dst)))]
177impl<M> ActorRef<M> {
178	#[inline]
179	pub(crate) fn from_wasm_inner(
180		processor: Rc<RefCell<Option<Box<dyn FnMut(M)>>>>,
181		alive: Arc<AtomicBool>,
182		queue: Rc<RefCell<Vec<M>>>,
183		processing: Rc<Cell<bool>>,
184	) -> Self {
185		Self {
186			inner: wasm::ActorRefInner::new(processor, alive, queue, processing),
187		}
188	}
189
190	#[inline]
191	pub fn send(&self, msg: M) -> Result<(), SendError<M>> {
192		self.inner.send(msg)
193	}
194
195	#[inline]
196	pub fn send_blocking(&self, msg: M) -> Result<(), SendError<M>> {
197		self.inner.send_blocking(msg)
198	}
199
200	#[inline]
201	pub fn is_alive(&self) -> bool {
202		self.inner.is_alive()
203	}
204
205	#[inline]
206	pub(crate) fn mark_stopped(&self) {
207		self.inner.mark_stopped()
208	}
209
210	#[inline]
211	pub(crate) fn processor(&self) -> &Rc<RefCell<Option<Box<dyn FnMut(M)>>>> {
212		&self.inner.processor
213	}
214}
215
216use std::error;
217#[cfg(not(reifydb_single_threaded))]
218use std::sync;
219
220#[cfg(not(reifydb_single_threaded))]
221use crossbeam_channel::Sender;
222#[cfg(reifydb_dst)]
223pub(crate) use dst::create_mailbox as create_dst_mailbox;
224#[cfg(not(reifydb_single_threaded))]
225pub(crate) use host::create_mailbox;
226#[cfg(all(reifydb_single_threaded, not(reifydb_dst)))]
227pub(crate) use wasm::create_actor_ref;
reifydb_runtime/actor/mailbox/mod.rs

reifydb_runtime/actor/mailbox/
mod.rs
