jsonrpc: a simple JSONRPC2 server

src/bin/cli.rs

@@ -14,7 +14,7 @@ use std::os::unix::net::UnixStream;
 // Exits with error
 fn show_usage() {
     eprintln!("Usage:");
-    eprintln!(" revault-cli [--conf conf_path] [--raw] <command> [<param 1> <param 2> ...]");
+    eprintln!(" minisafe-cli [--conf conf_path] [--raw] <command> [<param 1> <param 2> ...]");
     process::exit(1);
 }
 
@@ -73,7 +73,7 @@ fn rpc_request(method: String, params: Vec<String>) -> Json {
     object.insert("jsonrpc".to_string(), Json::String("2.0".to_string()));
     object.insert(
         "id".to_string(),
-        Json::String(format!("revault-cli-{}", process::id())),
+        Json::String(format!("minisafe-cli-{}", process::id())),
     );
     object.insert("method".to_string(), method);
     object.insert("params".to_string(), params);
@@ -94,7 +94,7 @@ fn socket_file(conf_file: Option<PathBuf>) -> PathBuf {
     [
         data_dir,
         config.bitcoind_config.network.to_string().as_str(),
-        "revaultd_rpc",
+        "minisafed_rpc",
     ]
     .iter()
     .collect()
@@ -127,7 +127,7 @@ fn main() {
         process::exit(1);
     });
     socket
-        .write_all(request.to_string().as_bytes())
+        .write_all(&[request.to_string().as_bytes(), b"\n"].concat())
         .unwrap_or_else(|e| {
             eprintln!("Writing to {:?}: '{}'", &socket_file, e);
             process::exit(1);
@@ -160,7 +160,7 @@ fn main() {
                         println!("{:#}", serde_json::json!({ "error": e }));
                     } else {
                         log::warn!(
-                            "revaultd response doesn't contain result or error: '{}'",
+                            "minisafed response doesn't contain result or error: '{}'",
                             response
                         );
                         println!("{:#}", response);

src/bin/daemon.rs

@@ -63,7 +63,9 @@ fn main() {
         // The panic hook will log::error
         panic!("Starting Minisafe daemon: {}", e);
     });
-    daemon.shutdown();
+    daemon
+        .rpc_server()
+        .expect("JSONRPC server must terminate cleanly");
 
     // We are always logging to stdout, should it be then piped to the log file (if self) or
     // not. So just make sure that all messages were actually written.

src/config.rs

@@ -94,6 +94,15 @@ pub struct Config {
     pub bitcoind_config: BitcoindConfig,
 }
 
+impl Config {
+    pub fn data_dir(&self) -> Option<PathBuf> {
+        self.data_dir
+            .as_ref()
+            .map(Clone::clone)
+            .or_else(config_folder_path)
+    }
+}
+
 #[derive(PartialEq, Eq, Debug)]
 pub enum ConfigError {
     DatadirNotFound,

src/jsonrpc/api.rs

@@ -0,0 +1,17 @@
+use crate::{
+    jsonrpc::{Error, Request, Response},
+    DaemonControl,
+};
+
+/// Handle an incoming JSONRPC2 request.
+pub fn handle_request(control: &DaemonControl, req: Request) -> Result<Response, Error> {
+    let result = match req.method.as_str() {
+        "getinfo" => serde_json::json!(&control.get_info()),
+        "getnewaddress" => serde_json::json!(&control.get_new_address()),
+        _ => {
+            return Err(Error::method_not_found());
+        }
+    };
+
+    Ok(Response::success(req.id, result))
+}

src/jsonrpc/mod.rs

@@ -0,0 +1,161 @@
+mod api;
+pub mod server;
+
+use std::{error, fmt};
+
+use serde::{self, Deserialize, Deserializer, Serialize, Serializer};
+
+#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+#[serde(untagged)]
+pub enum Params {
+    Array(Vec<serde_json::Value>),
+    Map(serde_json::Map<String, serde_json::Value>),
+}
+
+#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+#[serde(untagged)]
+pub enum ReqId {
+    Num(u64),
+    Str(String),
+}
+
+/// A JSONRPC2 request. See https://www.jsonrpc.org/specification#request_object.
+#[derive(Clone, Debug, PartialEq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Request {
+    /// Version. Must be "2.0".
+    pub jsonrpc: String,
+    /// Command name.
+    pub method: String,
+    /// Command parameters.
+    pub params: Option<Params>,
+    /// Request identifier.
+    pub id: ReqId,
+}
+
+/// JSONRPC2 error codes. See https://www.jsonrpc.org/specification#error_object.
+#[derive(Debug, PartialEq, Clone)]
+pub enum ErrorCode {
+    /// The method does not exist / is not available.
+    MethodNotFound,
+    /// Invalid method parameter(s).
+    InvalidParams,
+    /// Reserved for implementation-defined server-errors.
+    ServerError(i64),
+}
+
+impl Into<i64> for &ErrorCode {
+    fn into(self) -> i64 {
+        match self {
+            ErrorCode::MethodNotFound => -32601,
+            ErrorCode::InvalidParams => -32602,
+            ErrorCode::ServerError(code) => *code,
+        }
+    }
+}
+
+impl From<i64> for ErrorCode {
+    fn from(code: i64) -> ErrorCode {
+        match code {
+            -32601 => ErrorCode::MethodNotFound,
+            -32602 => ErrorCode::InvalidParams,
+            code => ErrorCode::ServerError(code),
+        }
+    }
+}
+
+impl<'a> Deserialize<'a> for ErrorCode {
+    fn deserialize<D>(deserializer: D) -> Result<ErrorCode, D::Error>
+    where
+        D: Deserializer<'a>,
+    {
+        let code: i64 = Deserialize::deserialize(deserializer)?;
+        Ok(code.into())
+    }
+}
+
+impl Serialize for ErrorCode {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_i64(self.into())
+    }
+}
+
+/// JSONRPC2 error response. See https://www.jsonrpc.org/specification#error_object.
+#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Error {
+    pub code: ErrorCode,
+    pub message: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub data: Option<serde_json::Value>,
+}
+
+impl Error {
+    pub fn new(code: ErrorCode, message: impl Into<String>) -> Error {
+        Error {
+            message: message.into(),
+            code,
+            data: None,
+        }
+    }
+
+    pub fn method_not_found() -> Error {
+        Error::new(ErrorCode::MethodNotFound, "Method not found")
+    }
+
+    pub fn invalid_params<M>(message: impl Into<String>) -> Error {
+        Error::new(
+            ErrorCode::InvalidParams,
+            format!("Invalid params: {}", message.into()),
+        )
+    }
+}
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        let code: i64 = (&self.code).into();
+        write!(f, "{}: {}", code, self.message)
+    }
+}
+
+impl error::Error for Error {}
+
+/// JSONRPC2 response. See https://www.jsonrpc.org/specification#response_object.
+#[derive(Clone, Debug, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct Response {
+    /// Version. Must be "2.0".
+    jsonrpc: String,
+    /// Required on success. Must not exist on error.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    result: Option<serde_json::Value>,
+    /// Required on error. Must not exist on success.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    error: Option<Error>,
+    /// Request identifier.
+    id: ReqId,
+}
+
+impl Response {
+    fn new(id: ReqId, result: Option<serde_json::Value>, error: Option<Error>) -> Response {
+        Response {
+            jsonrpc: "2.0".to_string(),
+            result,
+            error,
+            id,
+        }
+    }
+
+    pub fn success(id: ReqId, result: serde_json::Value) -> Response {
+        Response::new(id, Some(result), None)
+    }
+
+    pub fn error(id: ReqId, error: Error) -> Response {
+        Response::new(id, None, Some(error))
+    }
+}

src/jsonrpc/server.rs

@@ -0,0 +1,380 @@
+//! JSONRPC2 server
+//!
+//! This module implements the connections and streams handling logic for receiving
+//! JSONRPC2 requests on a Unix Domain Socket.
+
+use crate::{
+    jsonrpc::{api, Request, Response},
+    DaemonControl,
+};
+
+use std::{
+    io,
+    os::unix::net,
+    path,
+    sync::{self, atomic},
+    thread, time,
+};
+
+// Maximum number of concurrent RPC connections we may accept.
+const MAX_CONNECTIONS: u32 = 16;
+
+// Read a command from the stream.
+//
+// In order to both treat commands separately (respond as soon as we read one), and support
+// multiple commands in a single read or in multiple parts, we are given the context as writable
+// arguments:
+//   - `buf` is the buffer used to read from the socket. It will be extended as needed. It must be
+//   initialized.
+//   - `end`: The index of the end of the data read from the stream. Since `buf` needs to be
+//   initialized with dummy values, it can be very different from `buf.len()`. Used to not check
+//   for the separator character in the parts of the buffer with dummy values.
+//   - `cursor`: The index at which we checked for the separator character (`\n`). Used to not
+//   check twice for it on the same buffer chunk.
+fn read_command(
+    stream: &mut dyn io::Read,
+    buf: &mut Vec<u8>,
+    end: &mut usize,
+    cursor: &mut usize,
+) -> Result<Option<Request>, io::Error> {
+    assert!(!buf.is_empty());
+
+    loop {
+        // First off, check if there are no existing commands in the buffer.
+        let pos = buf[*cursor..*end].iter().position(|byt| byt == &b'\n');
+        log::trace!(
+            "pos: {:?}, buf[cur..end]: {:?}",
+            pos,
+            String::from_utf8_lossy(&buf[*cursor..*end])
+        );
+        if let Some(pos) = pos {
+            log::trace!(
+                "Parsing Request from: {:?}",
+                String::from_utf8_lossy(&buf[..*cursor + pos])
+            );
+            // TODO: don't return an io::Error here, instead try to parse a Request. Failing that,
+            // try to parse a serde_json::Value. Then return accordingly a JSONRPC "malformed
+            // request" or "invalid JSON" error.
+            let req: Request = serde_json::from_slice(&buf[..*cursor + pos])?;
+            *buf = buf[pos + 1..].to_vec(); // FIXME: can we avoid reallocating here?
+            *cursor = 0;
+            *end -= pos + 1;
+
+            return Ok(Some(req));
+        }
+
+        // If nothing can be gathered from the buffer, continue reading.
+        let new_read = stream.read(&mut buf[*end..])?;
+        if new_read == 0 {
+            return Ok(None);
+        }
+
+        // If we filled the buffer, increase its size and try again.
+        *end += new_read;
+        let buffer_filled = *end == buf.len();
+        if buffer_filled {
+            buf.resize(buf.len() * 2, 0);
+            continue;
+        }
+    }
+}
+
+// Handle all messages from this connection.
+fn connection_handler(
+    control: sync::Arc<sync::Mutex<DaemonControl>>,
+    mut stream: net::UnixStream,
+) -> Result<(), io::Error> {
+    let mut buf = vec![0; 2048];
+    let mut end = 0;
+    let mut cursor = 0;
+
+    loop {
+        let req = match read_command(&mut stream, &mut buf, &mut end, &mut cursor)? {
+            Some(req) => req,
+            None => {
+                // Connection closed.
+                return Ok(());
+            }
+        };
+
+        // TODO: respond in case of invalid JSON or invalid JSONRPC request.
+        let req_id = req.id.clone();
+
+        log::trace!("JSONRPC request: {:?}", serde_json::to_string(&req));
+        let response = api::handle_request(&control.lock().unwrap(), req)
+            .unwrap_or_else(|e| Response::error(req_id, e));
+        log::trace!("JSONRPC response: {:?}", serde_json::to_string(&response));
+        if let Err(e) = serde_json::to_writer(&stream, &response) {
+            log::error!("Error writing response: '{}'", e);
+            return Ok(());
+        }
+    }
+}
+
+// FIXME: have a decent way to share the DaemonControl between connections. Maybe make it Clone?
+/// The main event loop. Wait for connections, and treat requests sent through them.
+pub fn rpcserver_loop(
+    listener: net::UnixListener,
+    daemon_control: sync::Arc<sync::Mutex<DaemonControl>>,
+) -> Result<(), io::Error> {
+    // Keep it simple. We don't need great performances so just treat each connection in
+    // its thread, with a given maximum number of connections.
+    let connections_counter = sync::Arc::from(atomic::AtomicU32::new(0));
+
+    loop {
+        let (connection, _) = match listener.accept() {
+            Ok(c) => c,
+            Err(e) => {
+                log::error!("Accepting new connection: '{}'", e);
+                continue;
+            }
+        };
+        log::trace!("New JSONRPC connection");
+
+        while connections_counter.load(atomic::Ordering::Relaxed) >= MAX_CONNECTIONS {
+            thread::sleep(time::Duration::from_millis(50));
+        }
+        connections_counter.fetch_add(1, atomic::Ordering::Relaxed);
+
+        let handler_id = connections_counter.load(atomic::Ordering::Relaxed);
+        thread::Builder::new()
+            .name(format!("minisafe-jsonrpc-{}", handler_id))
+            .spawn({
+                let control = daemon_control.clone();
+                let counter = connections_counter.clone();
+
+                move || {
+                    if let Err(e) = connection_handler(control, connection) {
+                        log::error!("Error while handling connection {}: '{}'", handler_id, e);
+                    } else {
+                        log::trace!("Connection {} terminated without error.", handler_id);
+                    }
+                    counter.fetch_sub(1, atomic::Ordering::Relaxed);
+                }
+            })?;
+    }
+}
+
+// Tries to bind to the socket, if we are told it's already in use try to connect
+// to check there is actually someone listening and it's not a leftover from a
+// crash.
+fn bind(socket_path: &path::Path) -> Result<net::UnixListener, io::Error> {
+    match net::UnixListener::bind(socket_path) {
+        Ok(l) => Ok(l),
+        Err(e) => {
+            if e.kind() == io::ErrorKind::AddrInUse {
+                return match net::UnixStream::connect(socket_path) {
+                    Ok(_) => Err(e),
+                    Err(_) => {
+                        // Ok, no one's here. Just delete the socket and bind.
+                        log::debug!("Removing leftover rpc socket.");
+                        std::fs::remove_file(socket_path)?;
+                        net::UnixListener::bind(socket_path)
+                    }
+                };
+            }
+
+            Err(e)
+        }
+    }
+}
+
+/// Bind to the UDS at `socket_path`
+pub fn rpcserver_setup(socket_path: &path::Path) -> Result<net::UnixListener, io::Error> {
+    log::debug!("Binding socket at {}", socket_path.display());
+    // Create the socket with RW permissions only for the user
+    let old_umask = unsafe { libc::umask(0o177) };
+    let listener = bind(&socket_path);
+    unsafe {
+        libc::umask(old_umask);
+    }
+
+    listener
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::jsonrpc::{Params, ReqId};
+
+    use std::{env, fs, io::Write, process};
+
+    fn read_one_command(socket_path: &path::Path) -> thread::JoinHandle<Option<Request>> {
+        let listener = rpcserver_setup(socket_path).unwrap();
+        thread::spawn(move || {
+            let (mut conn, _) = listener.accept().unwrap();
+            let mut buf = vec![0; 32];
+            let mut end = 0;
+            let mut cursor = 0;
+            read_command(&mut conn, &mut buf, &mut end, &mut cursor).unwrap()
+        })
+    }
+
+    fn read_all_commands(socket_path: &path::Path) -> thread::JoinHandle<Vec<Request>> {
+        let listener = rpcserver_setup(socket_path).unwrap();
+        thread::spawn(move || {
+            let (mut conn, _) = listener.accept().unwrap();
+            let mut buf = vec![0; 32];
+            let mut end = 0;
+            let mut cursor = 0;
+            let mut reqs = Vec::new();
+
+            loop {
+                match read_command(&mut conn, &mut buf, &mut end, &mut cursor).unwrap() {
+                    Some(req) => {
+                        reqs.push(req);
+                    }
+                    None => return reqs,
+                }
+            }
+        })
+    }
+
+    fn write_messages(socket_path: &path::Path, messages: &[&[u8]]) {
+        let mut client = net::UnixStream::connect(&socket_path).unwrap();
+        for mess in messages {
+            client.write_all(&mess).unwrap();
+            // Simulate throttling, this mimics real conditions and actually triggered a crash.
+            thread::sleep(time::Duration::from_millis(50));
+        }
+    }
+
+    #[test]
+    fn command_read_single() {
+        let socket_path = env::temp_dir().join(format!(
+            "minisafed-jsonrpc-socket-{}-{:?}",
+            process::id(),
+            thread::current().id()
+        ));
+
+        // A simple dummy request
+        let t = read_all_commands(&socket_path);
+        let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": {"a": "b"}}"#;
+        let parsed_req: Request = serde_json::from_slice(req).unwrap();
+        write_messages(&socket_path, &[req, b"\n"]);
+        let read_req = t.join().unwrap();
+        assert_eq!(parsed_req, read_req[0]);
+
+        // Same, but with params as a list and a string id
+        let t = read_one_command(&socket_path);
+        let req = br#"{"jsonrpc": "2.0", "id": "987-abc", "method": "test", "params": ["a", 10]}"#;
+        let parsed_req: Request = serde_json::from_slice(req).unwrap();
+        write_messages(&socket_path, &[req, b"\n"]);
+        let read_req = t.join().unwrap().unwrap();
+        assert_eq!(parsed_req, read_req);
+
+        fs::remove_file(&socket_path).unwrap();
+    }
+
+    #[test]
+    fn command_read_parts() {
+        let socket_path = env::temp_dir().join(format!(
+            "minisafed-jsonrpc-socket-{}-{:?}",
+            process::id(),
+            thread::current().id()
+        ));
+
+        // A single request written in two parts
+        let t = read_one_command(&socket_path);
+        let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": ["a", 10]}"#;
+        let parsed_req: Request = serde_json::from_slice(req).unwrap();
+        write_messages(
+            &socket_path,
+            &[&req[..req.len() / 2], &req[req.len() / 2..], b"\n"],
+        );
+        let read_req = t.join().unwrap().unwrap();
+        assert_eq!(parsed_req, read_req);
+
+        // A single request written in many parts
+        let t = read_one_command(&socket_path);
+        let req = br#"{"jsonrpc": "2.0", "id": 0, "method": "test", "params": ["a", 10]}"#;
+        let parsed_req: Request = serde_json::from_slice(req).unwrap();
+        let tmp: Vec<Vec<u8>> = req.into_iter().map(|c| vec![*c]).collect();
+        let mut to_send: Vec<&[u8]> = tmp.iter().map(|v| v.as_slice()).collect();
+        to_send.push(b"\n");
+        write_messages(&socket_path, &to_send);
+        let read_req = t.join().unwrap().unwrap();
+        assert_eq!(parsed_req, read_req);
+
+        fs::remove_file(&socket_path).unwrap();
+    }
+
+    #[test]
+    fn command_read_multiple() {
+        let socket_path = env::temp_dir().join(format!(
+            "minisafed-jsonrpc-socket-{}-{:?}",
+            process::id(),
+            thread::current().id()
+        ));
+
+        // Multiple requests, in parts
+        let t = read_all_commands(&socket_path);
+        let reqs = [
+            &br#"{"jsonrpc": "2.0", "id": 20478, "me"#[..],
+            br#"thod": "test", "params": ["a", 10]}"#,
+            b"\n",
+            br#"{"jsonrpc": "2.0", "id": 20479, "method": "testADZ", "params": {}}"#,
+            b"\n",
+            br#"{"jsonrpc": "2.0", "id": 20499, "method": "t"#,
+            br#"e_edzA", "params": {"ttt": 980}}"#,
+            b"\n",
+        ];
+        let parsed_reqs: Vec<Request> = vec![
+            serde_json::from_slice(&[reqs[0], reqs[1]].concat()).unwrap(),
+            serde_json::from_slice(reqs[3]).unwrap(),
+            serde_json::from_slice(&[reqs[5], reqs[6]].concat()).unwrap(),
+        ];
+        write_messages(&socket_path, &reqs);
+        let read_reqs = t.join().unwrap();
+        assert_eq!(parsed_reqs, read_reqs);
+
+        // The same requests, sent at once.
+        let t = read_all_commands(&socket_path);
+        let req_parts = [
+            &br#"{"jsonrpc": "2.0", "id": 20478, "method": "test", "params": ["a", 10]}"#[..],
+            b"\n",
+            br#"{"jsonrpc": "2.0", "id": 20479, "method": "testADZ", "params": {}}"#,
+            b"\n",
+            br#"{"jsonrpc": "2.0", "id": 20499, "method": "te_edzA", "params": {"ttt": 980}}"#,
+            b"\n",
+        ]
+        .concat();
+        write_messages(&socket_path, &[req_parts.as_slice()]);
+        let read_reqs = t.join().unwrap();
+        assert_eq!(parsed_reqs, read_reqs);
+
+        fs::remove_file(&socket_path).unwrap();
+    }
+
+    #[test]
+    fn command_read_linebreak() {
+        let socket_path = env::temp_dir().join(format!(
+            "minisafed-jsonrpc-socket-{}-{:?}",
+            process::id(),
+            thread::current().id()
+        ));
+
+        // Multiple requests, in parts
+        let t = read_one_command(&socket_path);
+        let mut params = serde_json::map::Map::new();
+        params.insert(
+            "dummy param".to_string(),
+            "dummy value
+        with line
+        breaks"
+                .to_string()
+                .into(),
+        );
+        let req = Request {
+            jsonrpc: "2.0".to_string(),
+            method: "dummy".to_string(),
+            params: Some(Params::Map(params)),
+            id: ReqId::Num(0),
+        };
+        write_messages(&socket_path, &[&serde_json::to_vec(&req).unwrap(), b"\n"]);
+        let read_req = t.join().unwrap().unwrap();
+        assert_eq!(req, read_req);
+
+        fs::remove_file(&socket_path).unwrap();
+    }
+}

src/lib.rs

@@ -5,15 +5,19 @@ pub mod config;
 mod daemonize;
 mod database;
 pub mod descriptors;
+#[cfg(feature = "jsonrpc_server")]
+mod jsonrpc;
 
 pub use miniscript;
 
+#[cfg(feature = "jsonrpc_server")]
+use crate::jsonrpc::server::{rpcserver_loop, rpcserver_setup};
 use crate::{
     bitcoin::{
         d::{BitcoinD, BitcoindError},
         poller, BitcoinInterface,
     },
-    config::{config_folder_path, Config},
+    config::Config,
     database::{
         sqlite::{FreshDbOptions, SqliteDb, SqliteDbError},
         DatabaseInterface,
@@ -183,9 +187,8 @@ impl DaemonHandle {
 
         // First, check the data directory
         let mut data_dir = config
-            .data_dir
-            .clone()
-            .unwrap_or(config_folder_path().ok_or(StartupError::DefaultDataDirNotFound)?);
+            .data_dir()
+            .ok_or(StartupError::DefaultDataDirNotFound)?;
         data_dir.push(config.bitcoind_config.network.to_string());
         let fresh_data_dir = !data_dir.as_path().exists();
         if fresh_data_dir {
@@ -259,6 +262,39 @@ impl DaemonHandle {
         })
     }
 
+    /// Start the JSONRPC server and listen for incoming commands until we die.
+    /// Like DaemonHandle::shutdown(), this stops the Bitcoin poller at teardown.
+    #[cfg(feature = "jsonrpc_server")]
+    pub fn rpc_server(self) -> Result<(), io::Error> {
+        let DaemonHandle {
+            control,
+            bitcoin_poller: poller,
+        } = self;
+
+        let rpc_socket: path::PathBuf = [
+            control
+                .config
+                .data_dir()
+                .expect("Didn't fail at startup, must not now")
+                .as_path(),
+            path::Path::new(&control.config.bitcoind_config.network.to_string()),
+            path::Path::new("minisafed_rpc"),
+        ]
+        .iter()
+        .collect();
+        let listener = rpcserver_setup(&rpc_socket)?;
+        log::info!("JSONRPC server started.");
+
+        // FIXME: don't use a Mutex...
+        let control = sync::Arc::from(sync::Mutex::from(control));
+        rpcserver_loop(listener, control)?;
+        log::info!("JSONRPC server stopped.");
+
+        poller.stop();
+
+        Ok(())
+    }
+
     // NOTE: this moves out the data as it should not be reused after shutdown
     /// Shut down the Minisafe daemon.
     pub fn shutdown(self) {