aurpc 0.3.1

Asynchronous UDP RPCs
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

use std::collections::HashMap;

use async_std::{net::SocketAddr, sync::Mutex};
use futures::channel::oneshot;

use crate::message::RpcMessage;

/// A map for RPCs that have been sent and are awaiting for a response.
///
/// A tuple of a socket address and request id is used as a key for a
/// oneshot::Sender to send the response through.
#[derive(Default)]
pub struct AwaitingRequestMap {
    map: Mutex<MapState>,
}

#[derive(Default)]
struct MapState {
    current_rid: u16,
    map: HashMap<(SocketAddr, u16), oneshot::Sender<RpcMessage>>,
}

impl AwaitingRequestMap {
    /// Puts a record in the map representing the socket address the request was
    /// sent to and the channel through which to send a possible response
    /// through.
    ///
    /// Returns the next available request id.
    pub async fn put(
        &self,
        addr: SocketAddr,
        sender: oneshot::Sender<RpcMessage>,
    ) -> u16 {
        let mut map_state = self.map.lock().await;

        let mut rid = map_state.current_rid;
        while map_state.map.contains_key(&(addr, rid)) {
            rid += rid.wrapping_add(1);
        }
        map_state.current_rid = rid;

        map_state.map.insert((addr, rid), sender);
        rid
    }

    /// Pops a record from the map, allowing the response to be sent through the
    /// returned channel.
    ///
    /// If there is no such record, then None will be returned.
    pub async fn pop(
        &self,
        addr: SocketAddr,
        rid: u16,
    ) -> Option<oneshot::Sender<RpcMessage>> {
        let mut map_state = self.map.lock().await;
        map_state.map.remove(&(addr, rid))
    }
}

#[cfg(test)]
mod tests {
    use crate::awaiting::AwaitingRequestMap;
    use futures::channel::oneshot;

    #[async_std::test]
    async fn put_and_pop_returns_some() {
        let map = AwaitingRequestMap::default();
        let (sender, mut receiver) = oneshot::channel();

        let socket = "0.0.0.0:12345".parse().unwrap();
        let rid = map.put(socket, sender).await;

        let _ = map.pop(socket, rid).await.expect("empty entry");
        assert!(receiver.try_recv().is_err(), "channel not cancelled");
    }

    #[async_std::test]
    async fn just_pop_returns_none() {
        let map = AwaitingRequestMap::default();

        let socket = "0.0.0.0:12345".parse().unwrap();
        let rid = 0;

        assert!(map.pop(socket, rid).await.is_none(), "received something");
    }
}