speare 0.4.3

actor-like thin abstraction over tokio::task and flume channels
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

use flume::{Receiver, Sender};
use std::{fmt, time::Duration};
use tokio::time;

/// Represents a request sent to a `Actor`.
/// `Request` holds the data sent to a `Actor` and provides a channel to reply back to the sender.
pub struct Request<Req, Res> {
    data: Req,
    tx: Sender<Res>,
}

impl<Req, Res> fmt::Debug for Request<Req, Res>
where
    Req: fmt::Debug,
    Res: fmt::Debug,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Request")
            .field("data", &self.data)
            .field("tx", &self.tx)
            .finish()
    }
}

impl<Req, Res> Request<Req, Res> {
    /// Returns a reference to the data inside the `Request`.
    pub fn data(&self) -> &Req {
        &self.data
    }

    /// Sends a response back to the requester.
    pub fn reply(&self, res: Res) {
        let _ = self.tx.send(res);
    }
}

///`Response<Res>` is used to asynchronously wait for and retrieve the result of a `Request<Req, Res>` sent to a `Actor`.
pub struct Response<Res> {
    rx: Receiver<Res>,
}

impl<Res> fmt::Debug for Response<Res>
where
    Res: fmt::Debug,
{
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Response").field("rx", &self.rx).finish()
    }
}

#[derive(Debug)]
/// Represents a failure when waiting for a `Response<_>`
pub enum ReqErr {
    /// Request object was dropped before replying.
    Dropped,
    /// Timed out waiting for response.
    Timeout,
}

impl fmt::Display for ReqErr {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Dropped => write!(f, "Request object dropped before replying."),
            Self::Timeout => write!(f, "Timed out before receiving response."),
        }
    }
}

impl std::error::Error for ReqErr {}

impl<Res> Response<Res> {
    /// Asynchronously wait for a value from this channel,
    /// returning an error if the corresponding `Request<_,_>` has been dropped.
    pub async fn recv(self) -> Result<Res, ReqErr> {
        self.rx.recv_async().await.map_err(|_| ReqErr::Dropped)
    }

    /// Asynchronously wait for a value from this channel,
    /// returning an error if the given `Duration` elapses before
    /// receiving the expected response, or if the corresponding `Request<_,_>` has been dropped.
    pub async fn recv_timeout(self, dur: Duration) -> Result<Res, ReqErr> {
        time::timeout(dur, self.recv())
            .await
            .map_err(|_| ReqErr::Timeout)
            .and_then(|x| x)
    }
}

/// Creates a paired `Request<Req, Res>` and `Response<Res>` for communication between `speare` actors.
/// ## Example
pub fn req_res<Req, Res>(req: Req) -> (Request<Req, Res>, Response<Res>) {
    let (tx, rx) = flume::unbounded();
    (Request { data: req, tx }, Response { rx })
}