unicorn-rpc 0.0.1

Internal communication distribution abstraction for unicorn.
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

//! Internal communication distribution abstraction for `unicorn`.

extern crate nanomsg;
extern crate unicorn_messages;

pub use nanomsg::Protocol;

use nanomsg::{Socket, Endpoint, Result};
use unicorn_messages::{Msg, encode_bytes, decode};
use std::io::Read;

pub enum RPCType {
    Connect,
    Bind
}

/// Performs RPC-like data exchange over `nanomsg::Protocol::Rep` sockets.
pub struct RPC {
    socket: Socket,
    endpoint: Endpoint,
    socket_type: Protocol
}

impl RPC {
    fn create_socket(socktype: Protocol) -> Socket {
        match Socket::new(socktype) {
            Ok(s) => s,
            Err(e) => panic!("[rpc] Error creating {:?} socket. Error: {}", socktype, e),
        }
    }

    fn create_device_socket(socktype: Protocol) -> Socket {
        match Socket::new_for_device(socktype) {
            Ok(s) => s,
            Err(e) => panic!("[rpc] Error creating device socket for {:?}. Error: {}", socktype, e),
        }
    }

    fn connect(mut socket: Socket, addr: &str) -> (Socket, Endpoint) {
        let endpoint = match socket.connect(addr.as_ref()) {
            Ok(ce) => {
                ce
            }
            Err(e) => panic!("[rpc] Unable to connect to remote Rep endpoint: {}\nError: {}", &addr, e),
        };
        (socket, endpoint)
    }

    fn bind(mut socket: Socket, addr: &str) -> (Socket, Endpoint) {
        let endpoint = match socket.bind(addr.as_ref()) {
            Ok(ce) => {
                ce
            }
            Err(e) => panic!("[rpc] Unable to bind Rep endpoint: {}\nError: {}", &addr, e),
        };
        (socket, endpoint)
    }

    fn generate(socket: Socket, sock_type: Protocol, conn_type: RPCType, addr: &str) -> RPC {
        let (socket, endpoint) = match conn_type {
            RPCType::Connect => RPC::connect(socket, addr),
            RPCType::Bind => RPC::bind(socket, addr),
        };
        RPC {
            socket: socket,
            endpoint: endpoint,
            socket_type: sock_type
        }
    }

    /// Link two sockets using a `nanomsg`
    /// [`device`](http://nanomsg.org/v0.8/nn_device.3.HTML).
    pub fn link(&self, s2: &RPC) -> Result<()> {
        Ok(try!(Socket::device(&self.socket, &s2.socket)))
    }

    /// Create a new `RPC` instance
    pub fn new(sock_type: Protocol, conn_type: RPCType, addr: &str) -> RPC {
        let socket = RPC::create_socket(sock_type);
        RPC::generate(socket, sock_type, conn_type, addr)
    }

    /// Create a new `RPC` instance for use in `nanomsg` `device`
    pub fn new_for_device(sock_type: Protocol, conn_type: RPCType, addr: &str) -> RPC {
        let socket = RPC::create_device_socket(sock_type);
        RPC::generate(socket, sock_type, conn_type, addr)
    }

    /// Run a remote command by sending a `message` to remote socket
    /// and reading a reply for the operation from the remote socket.
    pub fn execute(&mut self, m: Msg) -> Result<Msg> {
        try!(self.send_msg(&m));
        let mut buff = String::new();
        try!(self.recv(&mut buff));
        match decode(&buff) {
            Some(msg) => Ok(msg),
            None => Ok(Msg::Error("Unable to decode message received from remote endpoint".to_string())),
        }
    }

    /// Receive from socket and write to given String
    pub fn recv(&mut self, buff: &mut String) -> Result<()> {
        try!(self.socket.read_to_string(buff));
        Ok(())
    }

    /// Send `Msg` on socket, after encoding `Msg` to byte vector.
    pub fn send_msg(&mut self, m: &Msg) -> Result<()> {
        try!(self.send(encode_bytes(m)));
        Ok(())
    }

    /// Send byte vector on socket
    pub fn send(&mut self, v: Vec<u8>) -> Result<()> {
        try!(self.socket.nb_write(v.as_slice()));
        Ok(())
    }

    /// Shutdown endpoint
    pub fn close(mut self) -> Result<()> {
        try!(self.endpoint.shutdown());
        Ok(())
    }

    /// Return socket type
    pub fn get_socket_type(&self) -> Protocol {
        self.socket_type
    }
}