dchat: pass command line arg to spin up new dchat instance

Cargo.lock

@@ -1221,7 +1221,7 @@ dependencies = [
  "subtle",
  "termion",
  "thiserror",
- "toml",
+ "toml 0.5.9",
  "tungstenite",
  "url",
  "wasmer",
@@ -1369,6 +1369,7 @@ dependencies = [
  "structopt",
  "structopt-toml",
  "termion",
+ "toml 0.4.10",
  "url",
 ]
 
@@ -2368,7 +2369,7 @@ dependencies = [
  "smol",
  "structopt",
  "structopt-toml",
- "toml",
+ "toml 0.5.9",
  "url",
 ]
 
@@ -3153,7 +3154,7 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1d6ea3c4595b96363c13943497db34af4460fb474a95c43f4446ad341b8c9785"
 dependencies = [
- "toml",
+ "toml 0.5.9",
 ]
 
 [[package]]
@@ -3163,7 +3164,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e17d47ce914bf4de440332250b0edd23ce48c005f59fab39d3335866b114f11a"
 dependencies = [
  "thiserror",
- "toml",
+ "toml 0.5.9",
 ]
 
 [[package]]
@@ -4075,7 +4076,7 @@ dependencies = [
  "skeptic",
  "structopt",
  "structopt-toml-derive",
- "toml",
+ "toml 0.5.9",
 ]
 
 [[package]]
@@ -4320,6 +4321,15 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
 
+[[package]]
+name = "toml"
+version = "0.4.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "758664fc71a3a69038656bee8b6be6477d2a6c315a6b81f7081f591bffa4111f"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "toml"
 version = "0.5.9"

example/dchat/Cargo.toml

@@ -31,5 +31,6 @@ termion = "1.5.6"
 serde = {version = "1.0.138", features = ["derive"]}
 structopt = "0.3.26"
 structopt-toml = "0.5.0"
+toml = "0.4.2"
 
 

example/dchat/src/main.rs

@@ -1,3 +1,4 @@
+use async_channel::{Receiver, Sender};
 use async_executor::Executor;
 use async_std::sync::Arc;
 use easy_parallel::Parallel;
@@ -6,24 +7,34 @@ use simplelog::WriteLogger;
 use std::{
     fs::File,
     io::{self, Read, Write},
+    path::PathBuf,
 };
+use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
 use termion::{async_stdin, event::Key, input::TermRead};
+use toml::Value;
 use url::Url;
 
 use darkfi::{
     net,
     net::Settings,
-    util::cli::{get_log_config, get_log_level},
+    util::{
+        cli::{get_log_config, get_log_level, spawn_config},
+        path::join_config_path,
+    },
     Result,
 };
 
 use crate::{dchatmsg::Dchatmsg, protocol_dchat::ProtocolDchat};
+use darkfi::net::settings::SettingsOpt;
 
 pub mod dchatmsg;
 pub mod protocol_dchat;
 
+const CONFIG_FILE_CONTENTS: &str = include_str!("../dchat_config.toml");
+
 struct Dchat {
     p2p: net::P2pPtr,
+    //msg_sub: net::MessageSubscription<Dchatmsg>,
 }
 
 impl Dchat {
@@ -32,7 +43,7 @@ impl Dchat {
     }
 
     async fn render(&self, ex: Arc<Executor<'_>>) -> Result<()> {
-        info!("DCHAT::render()::start");
+        info!(target: "dchat", "DCHAT::render()::start");
         let mut stdout = io::stdout().lock();
         let mut stdin = async_stdin();
 
@@ -47,7 +58,7 @@ impl Dchat {
             for k in stdin.by_ref().keys() {
                 match k.unwrap() {
                     Key::Char('q') => {
-                        info!("DCHAT::Q pressed.... exiting");
+                        info!(target: "dchat", "DCHAT::Q pressed.... exiting");
                         return Ok(())
                     }
                     Key::Char('i') => {}
@@ -72,43 +83,87 @@ impl Dchat {
         return Ok(input)
     }
 
-    async fn register_protocol(&self) -> Result<()> {
-        info!("DCHAT::register_protocol()::start");
+    async fn register_protocol(&self, p2p_send_channel: Sender<Dchatmsg>) -> Result<()> {
+        info!(target: "dchat", "dchat::register_protocol()::start");
         let registry = self.p2p.protocol_registry();
         registry
-            .register(net::SESSION_ALL, move |channel, p2p| async move {
-                ProtocolDchat::init(channel, p2p).await
+            .register(net::SESSION_ALL, move |channel, p2p| {
+                let sender = p2p_send_channel.clone();
+                async move { ProtocolDchat::init(channel, p2p, sender).await }
             })
             .await;
-        info!("DCHAT::register_protocol()::stop");
+        info!(target: "dchat", "DCHAT::register_protocol()::stop");
         Ok(())
     }
 
     async fn start(&self, ex: Arc<Executor<'_>>) -> Result<()> {
-        info!("DCHAT::start()::start");
+        info!(target: "dchat", "DCHAT::start()::start");
+
+        //let sub = net::MessageSubscription::new();
+        let (p2p_send_channel, p2p_recv_channel) = async_channel::unbounded::<Dchatmsg>();
+
         let ex2 = ex.clone();
+        let ex3 = ex.clone();
 
         let dchat = Dchat::new(self.p2p.clone());
 
-        dchat.register_protocol().await?;
+        dchat.register_protocol(p2p_send_channel).await?;
 
         self.p2p.clone().start(ex.clone()).await?;
         ex2.spawn(self.p2p.clone().run(ex.clone())).detach();
+        self.start_p2p_receive_loop(ex3, p2p_recv_channel);
 
-        info!("DCHAT::start()::stop");
+        info!(target: "dchat", "DCHAT::start()::stop");
         Ok(())
     }
 
+    fn start_p2p_receive_loop(
+        &self,
+        executor: Arc<Executor<'_>>,
+        p2p_receiver: Receiver<Dchatmsg>,
+    ) {
+        //let senders = self.senders.clone();
+        executor
+            .spawn(async move {
+                while let Ok(msg) = p2p_receiver.recv().await {
+                    info!(target: "dchat", "START P2P RECEIVE LOOP:: RECEIVED MSG {:?}", msg);
+                    //senders.notify(msg).await;
+                }
+            })
+            .detach();
+    }
+    //async fn send_loop(&self) -> Result<()> {
+    //    let dchatmsg = Dchatmsg { message };
+    //    loop {
+
+    //    }
+    //}
     async fn send(&self, message: String) -> Result<()> {
         let mut stdout = io::stdout().lock();
         stdout.write_all(b"sending: ")?;
         stdout.write_all(message.as_bytes())?;
         let dchatmsg = Dchatmsg { message };
-        self.p2p.broadcast(dchatmsg).await?;
+        match self.p2p.broadcast(dchatmsg).await {
+            Ok(_o) => {
+                info!(target: "dchat", "SEND: MSG BROADCAST SUCCESSFULLY");
+            }
+            Err(e) => {
+                error!(target: "dchat", "SEND: MSG FAILED TO BROADCAST {}", e);
+            }
+        }
         Ok(())
     }
 }
 
+fn get_settings_from_config(flag: String) -> Result<SettingsOpt> {
+    let name = format!("dchat_{}.toml", flag);
+    let cfg_path = join_config_path(&PathBuf::from(name))?;
+    spawn_config(&cfg_path, CONFIG_FILE_CONTENTS.as_bytes())?;
+    let config: SettingsOpt =
+        SettingsOpt::from_args_with_toml(&std::fs::read_to_string(cfg_path)?).unwrap();
+    Ok(config)
+}
+
 #[async_std::main]
 async fn main() -> Result<()> {
     let log_level = get_log_level(1);
@@ -122,22 +177,16 @@ async fn main() -> Result<()> {
     let inbound = Url::parse("tcp://127.0.0.1:55554").unwrap();
     let ext_addr = Url::parse("tcp://127.0.0.1:55544").unwrap();
 
-    let settings = Settings {
-        inbound: Some(inbound),
-        outbound_connections: 0,
-        manual_attempt_limit: 0,
-        seed_query_timeout_seconds: 8,
-        connect_timeout_seconds: 10,
-        channel_handshake_seconds: 4,
-        channel_heartbeat_seconds: 10,
-        outbound_retry_seconds: 1200,
-        external_addr: Some(ext_addr),
-        peers: Vec::new(),
-        seeds: vec![seed],
-        node_id: String::new(),
+    // TODO: error
+    let settings: SettingsOpt = match std::env::args()
+        .nth(1)
+        .expect("string identifier missing: please pass a string to name your dchat instance")
+    {
+        flag => get_settings_from_config(flag)?,
     };
 
-    let p2p = net::P2p::new(settings).await;
+    let p2p = net::P2p::new(settings.into()).await;
+
     let p2p = p2p.clone();
 
     let nthreads = num_cpus::get();

example/dchat/src/protocol_dchat.rs

@@ -2,7 +2,7 @@ use async_executor::Executor;
 use async_std::sync::Arc;
 use async_trait::async_trait;
 use darkfi::{net, Result};
-use log::info;
+use log::{error, info};
 
 use crate::dchatmsg::Dchatmsg;
 
@@ -10,10 +10,15 @@ pub struct ProtocolDchat {
     jobsman: net::ProtocolJobsManagerPtr,
     msg_sub: net::MessageSubscription<Dchatmsg>,
     p2p: net::P2pPtr,
+    p2p_send_channel: async_channel::Sender<Dchatmsg>,
 }
 
 impl ProtocolDchat {
-    pub async fn init(channel: net::ChannelPtr, p2p: net::P2pPtr) -> net::ProtocolBasePtr {
+    pub async fn init(
+        channel: net::ChannelPtr,
+        p2p: net::P2pPtr,
+        p2p_send_channel: async_channel::Sender<Dchatmsg>,
+    ) -> net::ProtocolBasePtr {
         info!(target: "dchat", "ProtocolDchat::init() [START]");
 
         let message_subsytem = channel.get_message_subsystem();
@@ -26,21 +31,42 @@ impl ProtocolDchat {
             jobsman: net::ProtocolJobsManager::new("ProtocolDchat", channel.clone()),
             msg_sub,
             p2p,
+            p2p_send_channel,
         })
     }
 
     async fn handle_receive_msg(self: Arc<Self>) -> Result<()> {
         info!(target: "dchat", "ProtocolDchat::handle_receive_msg() [START]");
-
         loop {
-            let msg = self.msg_sub.receive().await?;
+            match self.msg_sub.receive().await {
+                Ok(msg) => {
+                    info!(target: "dchat", "RECEIVED HANDLE_RECEIVE_MSG {:?}", msg);
+                    let msg = (*msg).to_owned();
+                    match self.p2p_send_channel.send(msg.clone()).await {
+                        Ok(o) => {
+                            info!(target: "dchat", "SENT MSG ACROSS p2p CHANNEL");
+                        }
+                        Err(e) => {
+                            error!(target: "dchat", "MSG SEND FAILED {}", e);
+                        }
+                    }
+                    info!(target: "dchat", "BROADCASTING MSG {:?}", msg);
+                    self.p2p.broadcast(msg).await?;
+                }
+                Err(e) => {
+                    error!(target: "dchat", "ERROR HANDLE_RECEIVE_MSG {:?}", e);
+                }
+            }
+        }
 
-            let msg = (*msg).to_owned();
+        //loop {
 
-            //self.sender.send(msg.clone()).await?;
+        //    //let msg = (*msg).to_owned();
 
-            self.p2p.broadcast(msg).await?;
-        }
+        //    //self.sender.send(msg.clone()).await?;
+
+        //    //self.p2p.broadcast(msg).await?;
+        //}
     }
 }