begin to add json rpc server to ircd

bin/ircd/Cargo.toml

@@ -9,8 +9,12 @@ clap = "2.34.0"
 smol = "1.2.5"
 simplelog = "0.11.1"
 async-std = "1.10.0"
+async-trait = "0.1.51"
 async-executor = "1.4.1"
 async-channel = "1.6.1"
 futures = "0.3.17"
 log = "0.4.14"
+rand = "0.8.4"
+serde_json = "1.0.72"
+serde = {version = "1.0.130", features = ["derive"]}
 

bin/ircd/src/irc_server.rs

@@ -1,14 +1,11 @@
 use std::{
     net::{TcpStream},
 };
-
-
-
+use rand::{RngCore, rngs::OsRng};
 use futures::{
     io::{WriteHalf}, AsyncWriteExt,
 };
 use log::{debug, info};
-
 use smol::Async;
 
 use drk::{
@@ -98,7 +95,9 @@ impl IrcServerConnection {
                 let message = &line[substr_idx + 1..];
                 info!("Message {}: {}", channel, message);
 
+
                 let protocol_msg = PrivMsg {
+                    id: OsRng.next_u32(),
                     nickname: self.nickname.clone(),
                     channel: channel.to_string(),
                     message: message.to_string(),

bin/ircd/src/main.rs

@@ -1,21 +1,29 @@
 #[macro_use]
 extern crate clap;
+use async_trait::async_trait;
 use std::{
     net::{SocketAddr, TcpListener, TcpStream},
     sync::Arc,
 };
-
 use async_executor::Executor;
 use async_std::io::BufReader;
 use futures::{
     AsyncBufReadExt, AsyncReadExt, FutureExt,
 };
+use serde_json::{json, Value};
 use log::{debug, error, info, warn};
 use simplelog::{ColorChoice, LevelFilter, TermLogger, TerminalMode};
 use smol::Async;
 
 use drk::{
     net,
+    rpc::{
+        jsonrpc::{
+            error as jsonerr, request as jsonreq, response as jsonresp, send_raw_request,
+            ErrorCode::*, JsonRequest, JsonResult,
+        },
+        rpcserver::{listen_and_serve, RequestHandler, RpcServerConfig},
+    },
     Error, Result,
 };
 
@@ -102,7 +110,7 @@ async fn channel_loop(
 
         debug!("NEWCHANNEL");
 
-        let protocol_privmsg = ProtocolPrivMsg::new(channel, sender.clone()).await;
+        let protocol_privmsg = ProtocolPrivMsg::new(channel, sender.clone(), p2p.clone()).await;
         protocol_privmsg.start(executor.clone()).await;
     }
 }
@@ -143,6 +151,11 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     // so detach them as background processes.
     executor.spawn(channel_loop(p2p.clone(), sender, executor.clone())).detach();
 
+    let rpc_interface = Arc::new(JsonRpcInterface {});
+    executor.spawn(async move {
+        listen_and_serve(server_config, rpc_interface, executor).await
+    }).detach();
+
     loop {
         let (stream, peer_addr) = match listener.accept().await {
             Ok((s, a)) => (s, a),
@@ -159,6 +172,33 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     }
 }
 
+struct JsonRpcInterface {
+}
+
+#[async_trait]
+impl RequestHandler for JsonRpcInterface {
+    async fn handle_request(&self, req: JsonRequest, _executor: Arc<Executor<'_>>) -> JsonResult {
+        if req.params.as_array().is_none() {
+            return JsonResult::Err(jsonerr(InvalidParams, None, req.id))
+        }
+
+        debug!(target: "RPC", "--> {}", serde_json::to_string(&req).unwrap());
+
+        match req.method.as_str() {
+            Some("say_hello") => return self.say_hello(req.id, req.params).await,
+            Some(_) | None => return JsonResult::Err(jsonerr(MethodNotFound, None, req.id)),
+        }
+    }
+}
+
+impl JsonRpcInterface {
+    // --> {"method": "say_hello", "params": []}
+    // <-- {"result": "hello world"}
+    async fn say_hello(&self, id: Value, _params: Value) -> JsonResult {
+        JsonResult::Resp(jsonresp(json!("hello world"), id))
+    }
+}
+
 fn main() -> Result<()> {
     TermLogger::init(
         LevelFilter::Debug,

bin/ircd/src/privmsg.rs

@@ -4,8 +4,11 @@ use drk::{
     serial::{Decodable, Encodable}, Result,
 };
 
+pub type PrivMsgId = u32;
+
 #[derive(Debug, Clone)]
 pub struct PrivMsg {
+    pub id: PrivMsgId,
     pub nickname: String,
     pub channel: String,
     pub message: String,
@@ -20,6 +23,7 @@ impl net::Message for PrivMsg {
 impl Encodable for PrivMsg {
     fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
         let mut len = 0;
+        len += self.id.encode(&mut s)?;
         len += self.nickname.encode(&mut s)?;
         len += self.channel.encode(&mut s)?;
         len += self.message.encode(&mut s)?;
@@ -30,6 +34,7 @@ impl Encodable for PrivMsg {
 impl Decodable for PrivMsg {
     fn decode<D: io::Read>(mut d: D) -> Result<Self> {
         Ok(Self {
+            id: Decodable::decode(&mut d)?,
             nickname: Decodable::decode(&mut d)?,
             channel: Decodable::decode(&mut d)?,
             message: Decodable::decode(&mut d)?,

bin/ircd/src/protocol_privmsg.rs

@@ -1,5 +1,6 @@
 use std::{
     sync::Arc,
+    collections::HashSet,
 };
 use log::debug;
 use async_executor::Executor;
@@ -7,18 +8,21 @@ use drk::{
     net, Result,
 };
 
-use crate::privmsg::PrivMsg;
+use crate::privmsg::{PrivMsgId, PrivMsg};
 
 pub struct ProtocolPrivMsg {
     notify_queue_sender: async_channel::Sender<Arc<PrivMsg>>,
     privmsg_sub: net::MessageSubscription<PrivMsg>,
     jobsman: net::ProtocolJobsManagerPtr,
+    privmsg_ids: HashSet<PrivMsgId>,
+    p2p: net::P2pPtr,
 }
 
 impl ProtocolPrivMsg {
     pub async fn new(
         channel: net::ChannelPtr,
         notify_queue_sender: async_channel::Sender<Arc<PrivMsg>>,
+        p2p: net::P2pPtr,
     ) -> Arc<Self> {
         let message_subsytem = channel.get_message_subsystem();
         message_subsytem.add_dispatch::<PrivMsg>().await;
@@ -32,6 +36,8 @@ impl ProtocolPrivMsg {
             notify_queue_sender,
             privmsg_sub,
             jobsman: net::ProtocolJobsManager::new("PrivMsgProtocol", channel),
+            privmsg_ids: HashSet::new(),
+            p2p,
         })
     }
 
@@ -53,6 +59,14 @@ impl ProtocolPrivMsg {
                 privmsg
             );
 
+            // Do we already have this message?
+            if self.privmsg_ids.contains(&privmsg.id) {
+                continue
+            }
+            // If not then broadcast to everybody else
+            let privmsg_copy = (*privmsg).clone();
+            self.p2p.broadcast(privmsg_copy).await?;
+
             self.notify_queue_sender.send(privmsg).await.expect("notify_queue_sender send failed!");
         }
     }