Cap'n Proto RPC

main.rs

#![feature(path)]
extern crate capnp;
extern crate capnpc;
extern crate "capnp-rpc" as capnp_rpc;
mod raft_capnp {
    include!("./schema/raft_capnp.rs");
}

use std::thread;

use raft_capnp::raft as Raft;

use capnp::capability::{FromServer, Server};
use capnp::{serialize_packed, MessageBuilder, MallocMessageBuilder};
use capnp::io::WriteOutputStream;

use capnp_rpc::capability::{InitRequest, LocalClient, WaitForContent};
use capnp_rpc::ez_rpc::{EzRpcServer, EzRpcClient};

struct RaftImpl;
impl Raft::Server for RaftImpl {
    fn append_entries(&mut self, mut context: Raft::AppendEntriesContext) {
        let (params, mut results) = context.get();
        // let params = params.get_params();

        let term = params.get_term();
        let leader_id = params.get_leader_id();
        let prev_log_index = params.get_prev_log_index();
        let prev_log_term = params.get_prev_log_term();
        let entries = {
            let target = params.get_entries();
            let size = target.len();
            let mut entries = Vec::with_capacity(size as usize);
            for i in 0..size {
                entries.push(target.get(i).to_string());
            }
            entries
        };
        let leader_commit = params.get_leader_commit();

        println!("
            append_entries:
                term: {},
                leaderId: {},
                prevLogIndex: {},
                prevLogTerm: {},
                entries: {:?},
                leaderCommit: {}
        ", term, leader_id, prev_log_index, prev_log_term, entries, leader_commit);

        results.set_term(1u64);
        results.set_success(true);

        context.done();
    }
    fn request_vote(&mut self, mut context: Raft::RequestVoteContext) {
        let (params, mut results) = context.get();

        let term = params.get_term();
        let candidate_id = params.get_candidate_id();
        let last_log_index = params.get_last_log_index();
        let last_log_term = params.get_last_log_term();

        println!("
            request_vote:
                term: {},
                candidateId: {},
                lastLogIndex: {},
                lastLogTerm: {},
        ", term, candidate_id, last_log_index, last_log_term);

        results.set_term(1u64);
        results.set_vote_granted(true);

        context.done();
    }
}

fn main() {
    println!("Creating server.");
    let rpc_server = EzRpcServer::new("localhost:8080").unwrap();
    let raft_server = Box::new(Raft::ServerDispatch { server : Box::new(RaftImpl)}) as Box<Server+Send>;
    rpc_server.export_cap("raft", raft_server);
    thread::spawn(move || {
        rpc_server.serve();
    });

    println!("Creating client.");
    let mut rpc_client = EzRpcClient::new("localhost:8080").unwrap();
    let raft_client: Raft::Client = rpc_client.import_cap("raft");

    println!("Issuing append_entries.");
    {
        let mut request = raft_client.append_entries_request();
        let mut builder = request.init();
        builder.set_term(0u64);
        builder.set_leader_id(0u64);
        builder.set_prev_log_index(0u64);
        builder.set_prev_log_term(0u64);
        // Do entries in a sec.
        builder.set_leader_commit(0u64);
        // Entry creation is scoped.
        {
            let mut entries = builder.init_entries(2u32);
            entries.set(0, "Foo");
            entries.set(1, "Bar");
        }
        let mut promise = request.send();

        println!("Retrieving append_entries response.");
        let response = promise.wait().unwrap();
        let term = response.get_term();
        let success = response.get_success();

        println!("
            Term: {}
            Success: {}
        ", term, success)
    }

    println!("Issuing request_vote.");
    {
        let mut request = raft_client.request_vote_request();
        let mut builder = request.init();
        builder.set_term(0u64);
        builder.set_candidate_id(0u64);
        builder.set_last_log_index(0u64);
        builder.set_last_log_term(0u64);
        let mut promise = request.send();

        println!("Retrieving request_vote response.");
        let response = promise.wait().unwrap();
        let term = response.get_term();
        let vote_granted = response.get_vote_granted();

        println!("
            Term: {}
            voteGranted: {}
        ", term, vote_granted)
    }
}

output

➜  capntest git:(master) ✗ ./target/capntest
Creating server.
Creating client.
Issuing append_entries.
Retrieving append_entries response.

            append_entries:
                term: 0,
                leaderId: 0,
                prevLogIndex: 0,
                prevLogTerm: 0,
                entries: ["Foo", "Bar"],
                leaderCommit: 0


            Term: 1
            Success: true

Issuing request_vote.
Retrieving request_vote response.

            request_vote:
                term: 0,
                candidateId: 0,
                lastLogIndex: 0,
                lastLogTerm: 0,


            Term: 1
            voteGranted: true

raft.capnp

@0xf64213cd3ccb41d5;
# unique file ID, generated by `capnp id`

interface Raft {
    appendEntries @0 (term :UInt64,
                      leaderId :UInt64,
                      prevLogIndex :UInt64,
                      prevLogTerm :UInt64,
                      entries :List(Text),
                      leaderCommit :UInt64)
                      -> (term :UInt64, success :Bool);
    requestVote @1 (term :UInt64,
                    candidateId :UInt64,
                    lastLogIndex :UInt64,
                    lastLogTerm :UInt64)
                    -> (term :UInt64, voteGranted :Bool);
}