ircd: create entry point socket for bots and plugins

Cargo.lock

@@ -2155,8 +2155,8 @@ dependencies = [
 ]
 
 [[package]]
-name = "ircd2"
-version = "0.3.0"
+name = "ircd"
+version = "0.1.0"
 dependencies = [
  "async-channel",
  "async-executor",

bin/ircd/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
-name = "ircd2"
-version = "0.3.0"
+name = "ircd"
+version = "0.1.0"
 homepage = "https://dark.fi"
 description = "P2P IRC daemon"
 authors = ["darkfi <dev@dark.fi>"]

bin/ircd/ircd_config.toml

@@ -4,6 +4,9 @@
 ## IRC listen URL
 #irc_listen="127.0.0.1:11066"
 
+## Plugin listen URL
+#plugin_listen="127.0.0.1:11077"
+
 ## Sets Datastore Path
 #datastore="~/.config/ircd"
 

bin/ircd/src/main.rs

@@ -4,15 +4,8 @@ use async_std::{
 };
 use std::net::SocketAddr;
 
-use async_channel::Receiver;
+use async_channel::{Receiver, Sender};
 use async_executor::Executor;
-use easy_parallel::Parallel;
-use futures::{io::BufReader, AsyncBufReadExt, AsyncReadExt, FutureExt};
-use log::{debug, error, info, warn};
-use simplelog::{ColorChoice, TermLogger, TerminalMode};
-use smol::future;
-use structopt_toml::StructOptToml;
-
 use darkfi::{
     async_daemonize,
     raft::Raft,
@@ -23,6 +16,12 @@ use darkfi::{
     },
     Error, Result,
 };
+use easy_parallel::Parallel;
+use futures::{io::BufReader, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, FutureExt};
+use log::{debug, error, info, warn};
+use simplelog::{ColorChoice, TermLogger, TerminalMode};
+use smol::future;
+use structopt_toml::StructOptToml;
 
 pub mod privmsg;
 pub mod rpc;
@@ -36,17 +35,19 @@ use crate::{
     settings::{Args, CONFIG_FILE, CONFIG_FILE_CONTENTS},
 };
 
-pub type SeenMsgId = Arc<Mutex<Vec<u32>>>;
+pub type SeenMsgIds = Arc<Mutex<Vec<u32>>>;
 
-async fn process_user_input(
-    mut line: String,
-    peer_addr: SocketAddr,
-    conn: &mut IrcServerConnection,
-    sender: async_channel::Sender<Privmsg>,
-    seen_msg_id: SeenMsgId,
-) -> Result<()> {
+fn build_irc_msg(msg: &Privmsg) -> String {
+    debug!("ABOUT TO SEND: {:?}", msg);
+    let irc_msg =
+        format!(":{}!anon@dark.fi PRIVMSG {} :{}\r\n", msg.nickname, msg.channel, msg.message,);
+    irc_msg
+}
+
+fn clean_input(mut line: String, peer_addr: &SocketAddr) -> Result<String> {
     if line.is_empty() {
-        warn!("Received empty line from {}. Closing connection.", peer_addr);
+        warn!("Received empty line from {}. ", peer_addr);
+        warn!("Closing connection.");
         return Err(Error::ChannelStopped)
     }
 
@@ -55,9 +56,17 @@ async fn process_user_input(
     line.pop();
     line.pop();
 
-    debug!("Received '{}' from {}", line, peer_addr);
+    Ok(line)
+}
+
+async fn broadcast_msg(
+    irc_msg: String,
+    peer_addr: SocketAddr,
+    conn: &mut IrcServerConnection,
+) -> Result<()> {
+    info!("Send msg to IRC server '{}' from {}", irc_msg, peer_addr);
 
-    if let Err(e) = conn.update(line, sender, seen_msg_id).await {
+    if let Err(e) = conn.update(irc_msg).await {
         warn!("Connection error: {} for {}", e, peer_addr);
         return Err(Error::ChannelStopped)
     }
@@ -65,49 +74,82 @@ async fn process_user_input(
     Ok(())
 }
 
+async fn plugin_process(
+    recv_for_plugin: Receiver<String>,
+    send_from_plugin: Sender<String>,
+    stream: TcpStream,
+) -> Result<()> {
+    let peer_addr = stream.peer_addr()?.clone();
+    let (reader, mut writer) = stream.split();
+    let mut reader = BufReader::new(reader);
+
+    loop {
+        let mut line = String::new();
+        futures::select! {
+            irc_msg = recv_for_plugin.recv().fuse() => {
+                let irc_msg = irc_msg?;
+                info!("Plugin recv {}", irc_msg);
+                writer.write_all(irc_msg.as_bytes()).await?;
+            },
+            err = reader.read_line(&mut line).fuse() => {
+                if let Err(e) = err {
+                    warn!("Read line error: {}", e);
+                    return Ok(())
+                }
+                let irc_msg = clean_input(line, &peer_addr)?;
+                info!("Plugin send {}", irc_msg);
+                send_from_plugin.send(irc_msg).await?;
+            }
+        };
+    }
+}
+
 async fn process(
-    receiver: Receiver<Privmsg>,
+    raft_receiver: Receiver<Privmsg>,
     stream: TcpStream,
     peer_addr: SocketAddr,
-    sender: async_channel::Sender<Privmsg>,
-    seen_msg_id: SeenMsgId,
+    raft_sender: Sender<Privmsg>,
+    send_for_plugin: Sender<String>,
+    recv_from_plugin: Receiver<String>,
+    seen_msg_id: SeenMsgIds,
 ) -> Result<()> {
     let (reader, writer) = stream.split();
 
     let mut reader = BufReader::new(reader);
-    let mut conn = IrcServerConnection::new(writer);
+    let mut conn = IrcServerConnection::new(writer, seen_msg_id.clone(), raft_sender);
 
     loop {
         let mut line = String::new();
         futures::select! {
-            privmsg = receiver.recv().fuse() => {
+            privmsg = raft_receiver.recv().fuse() => {
+                info!("Receive msg from raft");
                 let msg = privmsg?;
 
                 let mut smi = seen_msg_id.lock().await;
                 if smi.contains(&msg.id) {
                    continue
                 }
-
                 smi.push(msg.id);
                 drop(smi);
 
-                debug!("ABOUT TO SEND: {:?}", msg);
-                let irc_msg = format!(":{}!anon@dark.fi PRIVMSG {} :{}\r\n",
-                                      msg.nickname,
-                                      msg.channel,
-                                      msg.message,
-                                      );
-
+                let irc_msg = build_irc_msg(&msg);
                 conn.reply(&irc_msg).await?;
+                send_for_plugin.send(irc_msg).await?;
             }
+            irc_msg = recv_from_plugin.recv().fuse() => {
+                let irc_msg = irc_msg?;
+                info!("Receive msg from plugin");
+                broadcast_msg(irc_msg, peer_addr, &mut conn).await?;
 
+            }
             err = reader.read_line(&mut line).fuse() => {
                 if let Err(e) = err {
                     warn!("Read line error. Closing stream for {}: {}", peer_addr, e);
                     return Ok(())
                 }
-
-                process_user_input(line, peer_addr, &mut conn, sender.clone(), seen_msg_id.clone()).await?;
+                info!("Receive msg from IRC server");
+                let irc_msg = clean_input(line, &peer_addr)?;
+                broadcast_msg(irc_msg, peer_addr,&mut conn).await?;
             }
         };
     }
@@ -117,29 +159,25 @@ async_daemonize!(realmain);
 async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let listener = TcpListener::bind(settings.irc_listen).await?;
     let local_addr = listener.local_addr()?;
-    info!("Listening on {}", local_addr);
-
-    let datastore_path = expand_path(&settings.datastore)?;
+    info!("IRC listening on {}", local_addr);
 
-    let seen_msg_id: SeenMsgId = Arc::new(Mutex::new(vec![]));
+    let seen_msg_id: SeenMsgIds = Arc::new(Mutex::new(vec![]));
 
-    let net_settings = settings.net;
     //
     //Raft
     //
+    let datastore_path = expand_path(&settings.datastore)?;
+    let net_settings = settings.net;
     let datastore_raft = datastore_path.join("ircd.db");
-
     let mut raft = Raft::<Privmsg>::new(net_settings.inbound, datastore_raft)?;
-
     let raft_sender = raft.get_broadcast();
-    let commits = raft.get_commits();
+    let raft_receiver = raft.get_commits();
 
     //
     // RPC interface
     //
     let rpc_config = RpcServerConfig {
         socket_addr: settings.rpc_listen,
-        // TODO: Use net/transport:
         use_tls: false,
         identity_path: Default::default(),
         identity_pass: Default::default(),
@@ -150,6 +188,9 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         listen_and_serve(rpc_config, rpc_interface, executor_cloned.clone()).await
     });
 
+    let (send_from_plugin, recv_from_plugin) = async_channel::unbounded::<String>();
+    let (send_for_plugin, recv_for_plugin) = async_channel::unbounded::<String>();
+
     //
     // IRC instance
     //
@@ -164,20 +205,55 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
                 }
             };
 
-            info!("Accepted client: {}", peer_addr);
+            info!("IRC Accepted client: {}", peer_addr);
 
             executor_cloned
                 .spawn(process(
-                    commits.clone(),
+                    raft_receiver.clone(),
                     stream,
                     peer_addr,
                     raft_sender.clone(),
+                    send_for_plugin.clone(),
+                    recv_from_plugin.clone(),
                     seen_msg_id.clone(),
                 ))
                 .detach();
         }
     });
 
+    //
+    // Plugin instance
+    //
+    let executor_cloned = executor.clone();
+    let send_from_plugin_cloned = send_from_plugin.clone();
+    let recv_for_plugin_cloned = recv_for_plugin.clone();
+    let plugin_task: smol::Task<Result<()>> = executor.spawn(async move {
+        if settings.plugin_listen.is_none() {
+            return Ok(())
+        }
+        let plugin_listener = TcpListener::bind(settings.plugin_listen.unwrap()).await?;
+        let plugin_local_addr = plugin_listener.local_addr()?;
+        info!("Plugin listening on {}", plugin_local_addr);
+        loop {
+            let (stream, peer_addr) = match plugin_listener.accept().await {
+                Ok((s, a)) => (s, a),
+                Err(e) => {
+                    error!("Failed listening for connections: {}", e);
+                    return Err(Error::ServiceStopped)
+                }
+            };
+
+            info!("Plugin Accepted client: {}", peer_addr);
+            executor_cloned
+                .spawn(plugin_process(
+                    recv_for_plugin_cloned.clone(),
+                    send_from_plugin_cloned.clone(),
+                    stream,
+                ))
+                .detach();
+        }
+    });
+
     let (signal, shutdown) = async_channel::bounded::<()>(1);
     ctrlc_async::set_async_handler(async move {
         warn!(target: "ircd", "ircd start Exit Signal");
@@ -185,6 +261,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         signal.send(()).await.unwrap();
         rpc_task.cancel().await;
         irc_task.cancel().await;
+        plugin_task.cancel().await;
     })
     .unwrap();
 

bin/ircd/src/rpc.rs

@@ -40,14 +40,4 @@ impl JsonRpcInterface {
     async fn pong(&self, id: Value, _params: Value) -> JsonResult {
         jsonrpc::response(json!("pong"), id).into()
     }
-
-    // TODO
-    // RPCAPI:
-    // Retrieves P2P network information.
-    // --> {"jsonrpc": "2.0", "method": "get_info", "params": [], "id": 42}
-    // <-- {"jsonrpc": "2.0", result": {"nodeID": [], "nodeinfo": [], "id": 42}
-    //async fn get_info(&self, id: Value, _params: Value) -> JsonResult {
-    //    let resp = self.p2p.get_info().await;
-    //    jsonrpc::response(resp, id).into()
-    //}
 }

bin/ircd/src/server.rs

@@ -5,7 +5,7 @@ use rand::{rngs::OsRng, RngCore};
 
 use darkfi::{Error, Result};
 
-use crate::privmsg::Privmsg;
+use crate::{privmsg::Privmsg, SeenMsgIds};
 
 pub struct IrcServerConnection {
     write_stream: WriteHalf<TcpStream>,
@@ -14,10 +14,16 @@ pub struct IrcServerConnection {
     is_registered: bool,
     nickname: String,
     _channels: Vec<String>,
+    seen_msg_id: SeenMsgIds,
+    p2p_sender: async_channel::Sender<Privmsg>,
 }
 
 impl IrcServerConnection {
-    pub fn new(write_stream: WriteHalf<TcpStream>) -> Self {
+    pub fn new(
+        write_stream: WriteHalf<TcpStream>,
+        seen_msg_id: SeenMsgIds,
+        p2p_sender: async_channel::Sender<Privmsg>,
+    ) -> Self {
         Self {
             write_stream,
             is_nick_init: false,
@@ -25,21 +31,18 @@ impl IrcServerConnection {
             is_registered: false,
             nickname: "".to_string(),
             _channels: vec![],
+            seen_msg_id,
+            p2p_sender,
         }
     }
 
-    pub async fn update(
-        &mut self,
-        line: String,
-        sender: async_channel::Sender<Privmsg>,
-        seen_msg_id: crate::SeenMsgId,
-    ) -> Result<()> {
+    pub async fn update(&mut self, line: String) -> Result<()> {
         let mut tokens = line.split_ascii_whitespace();
         // Commands can begin with :garbage but we will reject clients doing
         // that for now to keep the protocol simple and focused.
         let command = tokens.next().ok_or(Error::MalformedPacket)?;
 
-        debug!("Received command: {}", command);
+        info!("Received command: {}", command);
 
         match command {
             "USER" => {
@@ -93,11 +96,11 @@ impl IrcServerConnection {
                     message: message.to_string(),
                 };
 
-                let mut smi = seen_msg_id.lock().await;
+                let mut smi = self.seen_msg_id.lock().await;
                 smi.push(random_id);
                 drop(smi);
 
-                sender.send(protocol_msg).await?;
+                self.p2p_sender.send(protocol_msg).await?;
             }
             "QUIT" => {
                 // Close the connection

bin/ircd/src/settings.rs

@@ -23,6 +23,9 @@ pub struct Args {
     /// IRC listen URL
     #[structopt(long = "irc", default_value = "127.0.0.1:11066")]
     pub irc_listen: SocketAddr,
+    /// Plugin listen URL
+    #[structopt(long)]
+    pub plugin_listen: Option<SocketAddr>,
     /// Sets Datastore Path
     #[structopt(long, default_value = "~/.config/ircd")]
     pub datastore: String,