bpxe 0.2.1

Business Process eXecution Engine
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

//! # Mailbox testing harness
//!
use crate::test::*;
use async_trait::async_trait;
use bpxe_internal_macros as bpxe_im;
use tokio::sync::broadcast;

/// Message receiver
#[async_trait]
pub(crate) trait Receiver<M> {
    /// Message receiver error
    type Error;

    /// Asynchronously receive a message
    async fn recv(&mut self) -> Result<M, Self::Error>;

    /// Returns whether returned error signifies source closure
    fn is_closed(err: Self::Error) -> bool;
}

#[async_trait]
impl<M> Receiver<M> for broadcast::Receiver<M>
where
    M: Clone + Send,
{
    type Error = broadcast::error::RecvError;

    async fn recv(&mut self) -> Result<M, Self::Error> {
        (self as &mut broadcast::Receiver<M>).recv().await
    }

    fn is_closed(err: Self::Error) -> bool {
        match err {
            broadcast::error::RecvError::Closed => true,
            _ => false,
        }
    }
}

/// Message mailbox
pub(crate) struct Mailbox<M, R> {
    messages: Vec<M>,
    receiver: R,
}

impl<M, R> Mailbox<M, R> {
    pub fn new(receiver: R) -> Self {
        Self {
            messages: Default::default(),
            receiver,
        }
    }
}

impl<M, R> Mailbox<M, R>
where
    R: Receiver<M>,
{
    /// Returns if there was or will be a message that makes `predicate` function return true
    pub async fn receive<F>(&mut self, predicate: F) -> bool
    where
        F: Fn(&M) -> bool,
    {
        let mut received = false;
        self.messages = std::mem::replace(&mut self.messages, vec![])
            .into_iter()
            .filter(|message| {
                if predicate(&message) {
                    received = true;
                    false
                } else {
                    true
                }
            })
            .collect();
        if !received {
            loop {
                tokio::task::yield_now().await;
                match self.receiver.recv().await {
                    Ok(message) => {
                        if predicate(&message) {
                            received = true;
                            break;
                        } else {
                            self.messages.push(message);
                        }
                    }
                    Err(e) => {
                        if R::is_closed(e) {
                            break;
                        }
                    }
                }
            }
        }
        return received;
    }
}

#[bpxe_im::test]
async fn assert_receive() {
    let (sender, receiver) = broadcast::channel(16);
    let mut mailbox = Mailbox::new(receiver);
    sender.send(1u8).unwrap();
    sender.send(2u8).unwrap();
    assert!(mailbox.receive(|m| *m == 2u8).await);
    assert!(mailbox.receive(|m| *m == 1u8).await);
}

#[tokio::test]
async fn refute_receive() {
    let (sender, receiver) = broadcast::channel(16);
    let mut mailbox = Mailbox::new(receiver);
    sender.send(1u8).unwrap();
    sender.send(2u8).unwrap();
    assert!(expects_timeout(mailbox.receive(|m| *m == 3u8))
        .await
        .is_ok());
}