syncweb_core/sync/
actor.rs

1use std::{future::Future, panic::AssertUnwindSafe};
2
3use n0_future::FutureExt;
4use tokio::sync::{mpsc, oneshot};
5
6/// Marker used to spawn a dedicated asynchronous storage actor.
7#[derive(Clone, Copy, Debug, Default)]
8#[non_exhaustive]
9pub struct Actor;
10
11/// Handle for sending requests to an [`Actor`].
12pub struct ActorHandle<M, R> {
13    sender: mpsc::UnboundedSender<(M, oneshot::Sender<R>)>,
14}
15
16impl<M, R> Clone for ActorHandle<M, R> {
17    fn clone(&self) -> Self {
18        Self {
19            sender: self.sender.clone(),
20        }
21    }
22}
23
24impl<M, R> std::fmt::Debug for ActorHandle<M, R> {
25    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        formatter.debug_struct("ActorHandle").finish_non_exhaustive()
27    }
28}
29
30impl Actor {
31    /// Spawn an actor loop. Handler panics are isolated to the request and do
32    /// not unwind the task or the caller.
33    #[must_use]
34    pub fn spawn<M, R, F, Fut>(handler: F) -> ActorHandle<M, R>
35    where
36        M: Send + 'static,
37        R: Default + Send + 'static,
38        F: Fn(M) -> Fut + Send + Sync + 'static,
39        Fut: Future<Output = R> + Send + 'static,
40    {
41        let (sender, mut receiver) = mpsc::unbounded_channel::<(M, oneshot::Sender<R>)>();
42        tokio::spawn(async move {
43            while let Some((message, response)) = receiver.recv().await {
44                let result = AssertUnwindSafe(handler(message)).catch_unwind().await;
45                if let Ok(value) = result {
46                    let _ = response.send(value);
47                } else {
48                    let _ = response.send(R::default());
49                }
50            }
51        });
52        ActorHandle { sender }
53    }
54}
55
56impl<M, R> ActorHandle<M, R>
57where
58    M: Send + 'static,
59    R: Default + Send + 'static,
60{
61    /// Send a message and await its response.
62    /// # Errors
63    ///
64    /// Returns an error if the actor has stopped or the response channel was dropped.
65    pub async fn request(&self, message: M) -> crate::Result<R> {
66        let (sender, receiver) = oneshot::channel();
67        self.sender
68            .send((message, sender))
69            .map_err(|error| crate::error::SyncwebError::operation("storage actor stopped", error))?;
70        receiver
71            .await
72            .map_err(|error| crate::error::SyncwebError::operation("storage actor response dropped", error))
73    }
74
75    #[must_use]
76    pub fn is_closed(&self) -> bool {
77        self.sender.is_closed()
78    }
79}
syncweb_core/sync/actor.rs

syncweb_core/sync/
actor.rs
