zestors 0.1.2

A fast and flexible actor-framework for building fault-tolerant Rust applications
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

use futures::{Future, FutureExt};
use std::{
    pin::Pin,
    task::{Context, Poll},
};
use tokio::sync::oneshot;
use crate::all::*;

/// Create a new request, consisting of a [`Tx<T>`] and an [`Rx<T>`].
/// The `Tx` (_transmitter_) can be used to send a single message `T` to the `Rx` (_receiver_).
///
/// This is just a wrapper around a [`tokio::sync::oneshot`] channel.
pub fn new_request<T>() -> (Tx<T>, Rx<T>) {
    let (tx, rx) = oneshot::channel();
    (Tx(tx), Rx(rx))
}

//------------------------------------------------------------------------------------------------
//  Tx
//------------------------------------------------------------------------------------------------

/// The transmitter part of a request, created with [`new_request`].
/// 
/// This implements [`MessageDerive<M>`] to be used with the [`derive@Message`] derive macro.
#[derive(Debug)]
pub struct Tx<M>(pub(super) oneshot::Sender<M>);

impl<M> Tx<M> {
    /// Send a message.
    pub fn send(self, msg: M) -> Result<(), TxError<M>> {
        self.0.send(msg).map_err(|msg| TxError(msg))
    }

    /// Whether the [`Rx`] has closed/dropped the oneshot-channel.
    pub fn is_closed(&self) -> bool {
        self.0.is_closed()
    }

    /// Wait for the [`Rx`] to close/drop the oneshot-channel.
    pub async fn closed(&mut self) {
        self.0.closed().await
    }
}

impl<M, R> MessageDerive<M> for Tx<R> {
    type Payload = (M, Rx<R>);
    type Returned = Tx<R>;

    fn create(msg: M) -> ((M, Rx<R>), Tx<R>) {
        let (tx, rx) = new_request();
        ((msg, rx), tx)
    }

    fn cancel(sent: (M, Rx<R>), _returned: Tx<R>) -> M {
        sent.0
    }
}

//------------------------------------------------------------------------------------------------
//  Rx
//------------------------------------------------------------------------------------------------

/// The receiver part of a request, created with [`new_request`].
/// 
/// This implements [`MessageDerive<M>`] to be used with the [`derive@Message`] derive macro.
#[derive(Debug)]
pub struct Rx<M>(pub(super) oneshot::Receiver<M>);

impl<M> Rx<M> {
    /// Attempt to take the message out, if it exists.
    pub fn try_recv(&mut self) -> Result<M, TryRxError> {
        self.0.try_recv().map_err(|e| e.into())
    }

    /// Block the thread while waiting for the message.
    pub fn recv_blocking(self) -> Result<M, RxError> {
        self.0.blocking_recv().map_err(|e| e.into())
    }

    /// Close the oneshot-channel, preventing the [`Tx`] from sending a message.
    pub fn close(&mut self) {
        self.0.close()
    }
}

impl<M, R> MessageDerive<M> for Rx<R> {
    type Payload = (M, Tx<R>);
    type Returned = Rx<R>;

    fn create(msg: M) -> ((M, Tx<R>), Rx<R>) {
        let (tx, rx) = new_request();
        ((msg, tx), rx)
    }

    fn cancel(sent: (M, Tx<R>), _returned: Rx<R>) -> M {
        sent.0
    }
}

impl<M> Unpin for Rx<M> {}

impl<M> Future for Rx<M> {
    type Output = Result<M, RxError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.0.poll_unpin(cx).map_err(|e| e.into())
    }
}

//------------------------------------------------------------------------------------------------
//  Errors
//------------------------------------------------------------------------------------------------

/// Error returned when receiving a message using an [`Rx`].
#[derive(Debug, Clone, Copy, PartialEq, PartialOrd, Eq, Ord, Hash, thiserror::Error)]
#[error("Failed to receive from Rx because it is closed.")]
pub struct RxError;

impl From<oneshot::error::RecvError> for RxError {
    fn from(_: oneshot::error::RecvError) -> Self {
        Self
    }
}

/// Error returned when trying to receive a message using an [`Rx`].
#[derive(Debug, Clone, Copy, PartialEq, PartialOrd, Eq, Ord, Hash, thiserror::Error)]
pub enum TryRxError {
    #[error("Closed")]
    Closed,
    #[error("Empty")]
    Empty,
}

impl From<oneshot::error::TryRecvError> for TryRxError {
    fn from(e: oneshot::error::TryRecvError) -> Self {
        match e {
            oneshot::error::TryRecvError::Empty => Self::Empty,
            oneshot::error::TryRecvError::Closed => Self::Closed,
        }
    }
}

/// Error returned when sending a message using a [`Tx`].
#[derive(Debug, Clone, Copy, PartialEq, PartialOrd, Eq, Ord, Hash, thiserror::Error)]
#[error("Failed to send to Tx because it is closed.")]
pub struct TxError<M>(pub M);