Tokio-RPC

An RPC framework for Rust base on tokio.

Frame Protocol

+-- MAGIC_VER: 1 --+--- request_id: 7 ---+-- payload_len: 4 --+-- payload --+ | 0b00101010, 42 | 0b00000000000001 | 0xffffffff, 4G - 1 | message | +------------------+---------------------+--------------------+-------------+

The min frame size: 1 + 7 + 4 = 12 bytes

The max frame size: 1 + 7 + 4 + 0xffffffff = 4294967307 bytes

Demo RPC with protobuf

https://github.com/iorust/tokio-rpc/blob/master/examples/protobuf_rpc.rs

Run:

sh cargo run --example protobuf_rpc

```rust extern crate tokiorpc; extern crate futures; extern crate tokiocore; extern crate tokio_service; extern crate protobuf;

use futures::{future, Future}; use tokiocore::reactor::Core; use tokioservice::{Service, NewService}; use std::{io, thread}; use std::time::Duration; use protobuf::core::{Message, parsefrombytes};

mod rpcpb;

struct Demo;

impl Service for Demo { type Request = Vec; type Response = Vec; type Error = io::Error; type Future = Box>;

fn call(&self, req: Self::Request) -> Self::Future {
    let req = parse_from_bytes::<rpcpb::Request>(req.as_slice());
    if let Err(err) = req {
        return future::err(io::Error::new(io::ErrorKind::Other, err)).boxed();
    }

    let req = req.unwrap();
    let req = req.get_cmd_ping_req();
    println!("Request {:?}", req);

    let mut msg = rpcpb::CmdPingResponse::new();
    msg.set_message(req.get_message().to_string());
    let mut res = rpcpb::Response::new();
    res.set_field_type(rpcpb::MessageType::CmdPing);
    res.set_cmd_ping_res(msg);

    match res.write_to_bytes() {
        Ok(val) => future::ok(val).boxed(),
        Err(err) => future::err(io::Error::new(io::ErrorKind::Other, err)).boxed(),
    }
}

}

impl NewService for Demo { type Request = Vec; type Response = Vec; type Error = io::Error; type Instance = Demo;

fn new_service(&self) -> io::Result<Self::Instance> {
    Ok(Demo {})
}

}

pub fn main() { let mut core = Core::new().unwrap(); let addr = "127.0.0.1:12345".parse().unwrap(); let demo = Demo {}; thread::spawn(move || { tokiorpc::serve(addr, demo); }); // A bit annoying, but we need to wait for the server to connect thread::sleep(Duration::frommillis(100));

let handle = core.handle();
core.run(tokio_rpc::Client::connect(&addr, &handle).and_then(|client| {
        let mut msg = rpcpb::CmdPingRequest::new();
        msg.set_message("Hello world!".to_string());

        let mut req = rpcpb::Request::new();
        req.set_field_type(rpcpb::MessageType::CmdPing);
        req.set_cmd_ping_req(msg);
        let buf = req.write_to_bytes().unwrap();
        client.call(buf)
            .and_then(|res| {
                          let res = parse_from_bytes::<rpcpb::Response>(res.as_slice());
                          println!("CLIENT Res: {:?}", res);
                          Ok(())
                      })
            .or_else(|err| {
                         println!("CLIENT Err: {:?}", err);
                         Ok(())
                     })
    }))
    .unwrap();

} ```