feat(raft): rpc handler

Author: Anton Melnikov

examples/cluster_node/src/lib.rs

@@ -6,7 +6,7 @@ use std::os::raw::c_int;
 use std::rc::{Rc, Weak};
 
 use tarantool::raft::Node;
-use tarantool::tuple::{FunctionArgs, FunctionCtx};
+use tarantool::tuple::{FunctionArgs, FunctionCtx, Tuple};
 
 #[derive(Default)]
 struct Global {
@@ -21,15 +21,12 @@ lazy_static! {
 }
 
 #[no_mangle]
-pub extern "C" fn run_node(_: FunctionCtx, _: FunctionArgs) -> c_int {
-    let node = Rc::new(
-        Node::new(
-            "libcluster_node.rpc",
-            vec!["127.0.0.1:3301", "127.0.0.1:3302", "127.0.0.1:3303"],
-            Default::default(),
-        )
-        .unwrap(),
-    );
+pub extern "C" fn run_node(_: FunctionCtx, args: FunctionArgs) -> c_int {
+    let args: Tuple = args.into();
+    let (bootstrap_addrs,) = args.into_struct::<(Vec<String>,)>().unwrap();
+
+    let node =
+        Rc::new(Node::new("libcluster_node.rpc", bootstrap_addrs, Default::default()).unwrap());
     GLOBAL.node.replace(Rc::downgrade(&node));
     node.run().unwrap();
     0

src/raft/mod.rs

@@ -10,9 +10,11 @@ use rand::random;
 use inner::{NodeAction, NodeInner};
 use net::{get_local_addrs, ConnectionPool};
 
-use crate::error::Error;
+use crate::error::{Error, TarantoolErrorCode};
+use crate::fiber::Cond;
 use crate::net_box::{Conn, ConnOptions, Options};
-use crate::tuple::{FunctionArgs, FunctionCtx};
+use crate::raft::inner::NodeEvent;
+use crate::tuple::{FunctionArgs, FunctionCtx, Tuple};
 
 mod fsm;
 pub mod inner;
@@ -24,6 +26,9 @@ pub struct Node {
     inner: RefCell<NodeInner>,
     connections: RefCell<ConnectionPool>,
     rpc_function: String,
+    events_cond: Cond,
+    events_buffer: RefCell<VecDeque<NodeEvent>>,
+    actions_buffer: RefCell<VecDeque<NodeAction>>,
     options: NodeOptions,
 }
 
@@ -67,17 +72,26 @@ impl Node {
             inner: RefCell::new(NodeInner::new(id, local_addrs, bootstrap_addrs_cfg)),
             connections: RefCell::new(ConnectionPool::new(options.connection_options.clone())),
             rpc_function: rpc_function.to_string(),
+            events_cond: Cond::new(),
+            events_buffer: RefCell::new(VecDeque::with_capacity(options.recv_queue_size)),
+            actions_buffer: RefCell::new(VecDeque::with_capacity(options.send_queue_size)),
             options,
         })
     }
 
     pub fn run(&self) -> Result<(), Error> {
-        let mut events = VecDeque::new();
-        let mut actions = VecDeque::new();
         loop {
-            self.inner.borrow_mut().update(&mut events, &mut actions);
-            for action in actions.drain(..) {
+            {
+                let mut actions = self.actions_buffer.borrow_mut();
+                let mut events = self.events_buffer.borrow_mut();
+                self.inner.borrow_mut().update(&mut events, &mut actions);
+            }
+
+            for action in self.actions_buffer.borrow_mut().drain(..) {
                 match action {
+                    NodeAction::Connect(id, addrs) => {
+                        self.connections.borrow_mut().connect(id, &addrs[..])?;
+                    }
                     NodeAction::Request(to, msg) => {
                         let mut conn_pool = self.connections.borrow_mut();
                         self.send(conn_pool.get(&to).unwrap(), rpc::Request::Bootstrap(msg))?;
@@ -89,11 +103,31 @@ impl Node {
                     _ => {}
                 };
             }
+
+            self.events_cond.wait();
         }
     }
 
     pub fn handle_rpc(&self, ctx: FunctionCtx, args: FunctionArgs) -> i32 {
-        unimplemented!();
+        let args: Tuple = args.into();
+
+        match args.into_struct::<rpc::Request>() {
+            Err(e) => set_error!(TarantoolErrorCode::Protocol, "{}", e),
+            Ok(request) => {
+                match request {
+                    rpc::Request::Bootstrap(msg) => {
+                        self.events_buffer
+                            .borrow_mut()
+                            .push_back(NodeEvent::Request(msg));
+                        self.events_cond.wait();
+                    }
+                    _ => unimplemented!(),
+                };
+
+                ctx.return_mp(&rpc::Response::Ack)
+                    .unwrap_or_else(|e| set_error!(TarantoolErrorCode::ProcC, "{}", e))
+            }
+        }
     }
 
     pub fn wait_ready(&self, timeout: Duration) -> Result<(), Error> {