rr-mux 0.1.0

A Request Response Multiplexer for building protocol handlers etc.
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



use std::sync::{Mutex, Arc};
use std::collections::HashMap;
use std::marker::PhantomData;


use futures::prelude::*;
use futures::sync::{oneshot, oneshot::Sender as OneshotSender};

use crate::connector::Connector;

/// Muxed is a container for either a request or response message
#[derive(Debug, PartialEq)]
pub enum Muxed<REQ, RESP> {
    Request(REQ),
    Response(RESP),
}

impl <REQ, RESP> Muxed<REQ, RESP> {
    /// Fetch a request if muxed contains a request type
    pub fn req(self) -> Option<REQ> {
        match self {
            Muxed::Request(req) => Some(req),
            _ => None
        }
    }

    /// Fetch a response if muxed contains a response type
    pub fn resp(self) -> Option<RESP> {
        match self {
            Muxed::Response(resp) => Some(resp),
            _ => None
        }
    }
}


/// Mux is a futures based request response multiplexer
/// ID is the request ID type
/// ADDR is the address for the REQ or RESP to be sent to
/// REQ and RESP are the request and response messages
/// SENDER is a futures-based sender for sending messages
pub struct Mux<ID, ADDR, REQ, RESP, ERR, SENDER> {
    requests: Arc<Mutex<HashMap<ID, Box<OneshotSender<RESP>>>>>,
    sender: Arc<Mutex<SENDER>>,
    _addr: PhantomData<ADDR>,
    _req: PhantomData<REQ>,
    _err: PhantomData<ERR>,
}

impl <ID, ADDR, REQ, RESP, ERR, SENDER> Mux <ID, ADDR, REQ, RESP, ERR, SENDER> 
where
    ID: std::cmp::Eq + std::hash::Hash + std::fmt::Debug + Clone + 'static,
    ADDR: std::fmt::Debug + 'static,
    REQ: std::fmt::Debug + 'static,
    RESP: std::fmt::Debug + 'static,
    ERR: std::fmt::Debug + 'static,
    SENDER: FnMut(ID, Muxed<REQ, RESP>, ADDR) -> Box<Future<Item=(), Error=ERR>> + 'static,
{
    /// Create a new mux over the provided sender
    pub fn new(sender: SENDER) -> Mux<ID, ADDR, REQ, RESP, ERR, SENDER> {
        Mux{
            requests: Arc::new(Mutex::new(HashMap::new())), 
            sender: Arc::new(Mutex::new(sender)),
            _addr: PhantomData,
            _req: PhantomData,
            _err: PhantomData,
        }
    }

    /// Handle a received message
    /// This either returns a pending response or passes request messages on
    pub fn handle(&mut self, id: ID, addr: ADDR, resp: Muxed<REQ, RESP>) -> Result<Option<(ADDR, REQ)>, ERR> {
        let r = match resp {
            // Requests get passed through the mux
            Muxed::Request(req) => {
                Some((addr, req))
            },
            // Responses get matched with outstanding requests
            Muxed::Response(resp) => {
                if let Some(ch) = self.requests.lock().unwrap().remove(&id) {
                    ch.send(resp).unwrap();
                } else {
                    info!("Response id: '{:?}', no request pending", id);
                }
                None
            }
        };

        Ok(r)
    }
}

impl <ID, ADDR, REQ, RESP, ERR, SENDER> Connector <ID, ADDR, REQ, RESP, ERR> for Mux <ID, ADDR, REQ, RESP, ERR, SENDER> 
where
    ID: std::cmp::Eq + std::hash::Hash + std::fmt::Debug + Clone + 'static,
    ADDR: std::fmt::Debug + 'static,
    REQ: std::fmt::Debug + 'static,
    RESP: std::fmt::Debug + 'static,
    ERR: std::fmt::Debug + 'static,
    SENDER: FnMut(ID, Muxed<REQ, RESP>, ADDR) -> Box<Future<Item=(), Error=ERR>> + 'static,
{

    /// Send and register a request
    fn request(&mut self, id: ID, addr: ADDR, req: REQ) -> Box<Future<Item=RESP, Error=ERR>> {
        // Create future channel
        let (tx, rx) = oneshot::channel();

        // Save response to map
        self.requests.lock().unwrap().insert(id.clone(), Box::new(tx));

        // Send request and return channel future
        let sender = self.sender.clone();
        Box::new(futures::lazy(move || {
            let sender = &mut *sender.lock().unwrap();
            (sender)(id, Muxed::Request(req), addr)
        })
        .and_then(|_| {
            // Panic on future closed, this is probably not desirable
            // TODO: fix
            rx.map_err(|_e| panic!() )
        }))
    }

    fn respond(&mut self, id: ID, addr: ADDR, resp: RESP) -> Box<Future<Item=(), Error=ERR>> {
        // Send request and return channel future
        let sender = self.sender.clone();
        Box::new(futures::lazy(move || {
            let sender = &mut *sender.lock().unwrap();
            (sender)(id, Muxed::Response(resp), addr)
        }))
    }
}