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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163


use std::net::{SocketAddr};
use std::time::Duration;

use futures::prelude::*;
use tokio::prelude::*;

use daemon_engine::{TcpConnection};
use daemon_engine::codecs::json::{JsonCodec};
use rr_mux::{Mux as BaseMux, Connector};

use crate::common::*;
use crate::manager::Manager;
use crate::error::Error;

pub mod spi;
use spi::Spi;
pub mod i2c;
use i2c::I2c;
pub mod pin;
use pin::Pin;

type Mux = BaseMux<u64, (), Request, Response, Error, ()>;


pub const TIMEOUT: Duration = Duration::from_secs(3);

/// Remote client for connecting to remote-hal server peripherals
/// 
/// THIS MUST BE RUN IN A MULTI-THREADED TOKIO CONTEXT
pub struct Client {
    connection: TcpConnection<JsonCodec<Request, Response, Error>>,
    mux: Mux,
}

unsafe impl Sync for Client {}
unsafe impl Send for Client {}

pub trait Requester {
    fn do_request(&mut self, path: &str, req: RequestKind) -> Box<Future<Item=ResponseKind, Error=Error> + Send + 'static>;
}

impl Requester for Mux {
    fn do_request(&mut self, path: &str, req: RequestKind) -> Box<Future<Item=ResponseKind, Error=Error> + Send + 'static> {
        let req = Request::new(path.to_owned(), req);
        info!("sending request {:?}", req);
        Box::new(self.request((), req.id, (), req)
        .timeout(TIMEOUT)
        .map_err(|e| e.into() )
        .then(|r| {
            let resp = match r {
                Err(e) => return Err(e),
                Ok(v) => v,
            };

            info!("received response {:?}", resp);

            match resp.0.kind {
                ResponseKind::Error(e) => Err(Error::Remote(e)),
                _ => Ok(resp.0.kind),
            }
        }))
    }
}

impl Client {
    /// Create a new remote-hal instance
    pub fn new(addr: SocketAddr) -> impl Future<Item=Self, Error=Error> {
        info!("client connecting to: {}", addr);

        TcpConnection::<JsonCodec<Request, Response, Error>>::new(&addr, JsonCodec::new()).map_err(|e| e.into() ).timeout(TIMEOUT).map_err(|e| e.into() ).map(|connection| {

            let (tx, rx) = connection.clone().split();

            info!("client connected");

            let mux = Mux::new();

            // Map mux output to tx
            let m = mux.clone();
            let tx_handle = tx.send_all(m.map(|(_req_id, _target, msg, _ctx)| msg.req().unwrap() ).map_err(|e| panic!(e) ));
            tokio::spawn(tx_handle.map(|_v| () ).map_err(|e| panic!(e) ));

            // Map rx to mux input
            let mut m = mux.clone();
            let rx_handle = rx.for_each(move |resp| m.handle_resp(resp.id, (), resp, ()) );
            tokio::spawn(rx_handle.map(|_v| () ).map_err(|e| panic!(e) ));
            
            Self{connection, mux}

        })
    }

    // Close a remote-hal instance
    pub fn close(self) {
        self.connection.close();
    }

    /// Pass-through for raw requests
    pub fn request(&mut self, device: &str, request: RequestKind) -> impl Future<Item=ResponseKind, Error=Error> {
        self.mux.do_request(device, request)
    }
}

impl Manager for Client {
    type Spi = Spi;
    type Pin = Pin;
    type I2c = I2c;

    /// Connect to a new Spi instance
    fn spi(&mut self, path: &str, baud: u32, mode: SpiMode) -> Box<Future<Item=Spi, Error=Error> + Send> {
        debug!("attempting connection to SPI device: {}", path);
        let device = path.to_owned();
        let mux = self.mux.clone();
        Box::new(self.mux.do_request(path, RequestKind::SpiConnect(SpiConnect{baud, mode}))
        .then(|res| {
            let resp = match res {
                Err(e) => return Err(e),
                Ok(r) => r,
            };
            match resp {
                ResponseKind::Ok => Ok(Spi::new(device, mux)),
                _ => Err(Error::InvalidResponse(resp)),
            }
        }))
    }

    /// Connect to a new Pin instance
    fn pin(&mut self, path: &str, mode: PinMode) -> Box<Future<Item=Pin, Error=Error> + Send> {
        debug!("attempting connection to Pin: {}", path);
        let device = path.to_owned();
        let mux = self.mux.clone();
        Box::new(self.mux.do_request(path, RequestKind::PinConnect(mode))
        .then(|res| {
            let resp = match res {
                Err(e) => return Err(e),
                Ok(r) => r,
            };
            match resp {
                ResponseKind::Ok => Ok(Pin::new(device, mux)),
                _ => Err(Error::InvalidResponse(resp)),
            }
        }))
    }

    /// Connect to a new I2c instance
    fn i2c(&mut self, path: &str) -> Box<Future<Item=I2c, Error=Error> + Send> {
        debug!("attempting connection to I2c: {}", path);
        let device = path.to_owned();
        let mux = self.mux.clone();
        Box::new(self.mux.do_request(path, RequestKind::I2cConnect)
        .then(|res| {
            let resp = match res {
                Err(e) => return Err(e),
                Ok(r) => r,
            };
            match resp {
                ResponseKind::Ok => Ok(I2c::new(device, mux)),
                _ => Err(Error::InvalidResponse(resp)),
            }
        }))
    }
}