job_abstrs 0.1.0

Abstractions for event-driven jobs.
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

use async_trait::async_trait;
use std::time::Duration;
use tokio::{task, time};

use crate::{McReceiver, MpSender, Run};

/// A job to whom resquests can be sent over a multiple producer channel and
/// from whom responses can be received over a multiple consumer channel.
#[async_trait]
pub trait SendReceive<JId, ReqTx, RespRx>: Run
where
    JId: Copy + Eq + Send + Sync + 'static,
    <Self as Run>::Error: Clone + Send + 'static,
    <Self as Run>::Request: Send + 'static,
    <Self as Run>::Response: Send + 'static,
    ReqTx: MpSender<Message = (JId, <Self as Run>::Request)> + Send + Sync + 'static,
    RespRx: McReceiver<Message = (JId, Result<<Self as Run>::Response, <Self as Run>::Error>)>
        + Send
        + Sync
        + 'static,
{
    /// Passes the request along with the job ID to the given [request_sender]
    /// and waits on a corresponding response from the given
    /// [response_receiver].
    async fn send_receive(
        request_sender: ReqTx,
        response_receiver: RespRx,
        jid: JId,
        req: <Self as Run>::Request,
        timeout: Duration,
    ) -> Result<<Self as Run>::Response, SendReceiveError<<Self as Run>::Error>> {
        // Start listenting for responses before the request is actually sent so
        // a response can't get sent before we start listening.
        let poller = task::spawn(async move {
            loop {
                match response_receiver.receive().await {
                    Ok((resp_job_id, resp_rslt)) if jid == resp_job_id => return resp_rslt,
                    _ => {}
                }
            }
        });

        request_sender
            .send((jid, req))
            .await
            .map_err(|_| SendReceiveError::RequestSender)?;
        let resp = time::timeout(timeout, poller)
            .await
            .map_err(|_| SendReceiveError::Elapsed)?
            .map_err(|_| SendReceiveError::Join)?
            .map_err(SendReceiveError::Job)?;
        Ok(resp)
    }
}

#[derive(thiserror::Error, Debug, Clone)]
pub enum SendReceiveError<JErr>
where
    JErr: Clone,
{
    #[error("error occured sending a request to the job processor")]
    RequestSender,
    #[error("error occured receiving a response from the job processor")]
    ResponseReceiver,
    #[error(transparent)]
    Job(JErr),
    #[error("error occured joining on threads")]
    Join,
    #[error("response was not received by within the expected timeout period")]
    Elapsed,
}