Merge branch 'dchat'

Cargo.lock

@@ -1221,7 +1221,7 @@ dependencies = [
  "subtle",
  "termion",
  "thiserror",
- "toml",
+ "toml 0.5.9",
  "tungstenite",
  "url",
  "wasmer",
@@ -1348,6 +1348,48 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "dchat"
+version = "0.1.0"
+dependencies = [
+ "async-channel",
+ "async-executor",
+ "async-std",
+ "async-trait",
+ "clap 3.2.8",
+ "darkfi",
+ "easy-parallel",
+ "futures",
+ "log",
+ "num_cpus",
+ "ringbuffer",
+ "serde",
+ "simplelog",
+ "smol",
+ "structopt",
+ "structopt-toml",
+ "termion",
+ "toml 0.4.10",
+ "url",
+]
+
+[[package]]
+name = "dchat_seed"
+version = "0.1.0"
+dependencies = [
+ "async-channel",
+ "async-executor",
+ "async-std",
+ "darkfi",
+ "easy-parallel",
+ "futures",
+ "log",
+ "num_cpus",
+ "simplelog",
+ "smol",
+ "url",
+]
+
 [[package]]
 name = "deflate"
 version = "0.8.6"
@@ -2251,7 +2293,7 @@ dependencies = [
  "smol",
  "structopt",
  "structopt-toml",
- "toml",
+ "toml 0.5.9",
  "url",
 ]
 
@@ -2948,7 +2990,7 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1d6ea3c4595b96363c13943497db34af4460fb474a95c43f4446ad341b8c9785"
 dependencies = [
- "toml",
+ "toml 0.5.9",
 ]
 
 [[package]]
@@ -2958,7 +3000,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e17d47ce914bf4de440332250b0edd23ce48c005f59fab39d3335866b114f11a"
 dependencies = [
  "thiserror",
- "toml",
+ "toml 0.5.9",
 ]
 
 [[package]]
@@ -3870,7 +3912,7 @@ dependencies = [
  "skeptic",
  "structopt",
  "structopt-toml-derive",
- "toml",
+ "toml 0.5.9",
 ]
 
 [[package]]
@@ -4117,6 +4159,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"

Cargo.toml

@@ -35,6 +35,9 @@ members = [
 	"src/sdk",
 	"src/util/derive",
 	"src/util/derive-internal",
+
+    "example/dchat",
+    "example/dchat-seed",
 ]
 
 [dependencies]

example/dchat-seed/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "dchat_seed"
+version = "0.1.0"
+edition = "2021"
+description = "Seed node for dchat"
+
+
+[dependencies]
+darkfi = {path = "../../", features = ["net", "rpc"]}
+
+# Async
+futures = "0.3.0"
+async-std = "1"
+async-executor = "1.4.1"
+async-channel = "1.6.1"
+easy-parallel = "3.2.0"
+smol = "1.2.5"
+num_cpus = "1.13.1"
+
+#Misc
+url = "2.2.2"
+log = "0.4.17"
+simplelog = "0.12.0"
+
+

example/dchat-seed/src/main.rs

@@ -0,0 +1,89 @@
+use async_executor::Executor;
+use async_std::sync::Arc;
+use easy_parallel::Parallel;
+//use log::{error, info, warn};
+use url::Url;
+
+use darkfi::{
+    net,
+    net::Settings,
+    util::cli::{get_log_config, get_log_level},
+    Result,
+};
+
+#[async_std::main]
+async fn main() -> Result<()> {
+    let log_level = get_log_level(1);
+    let log_config = get_log_config();
+
+    let env_log_file_path = match std::env::var("DARKFI_LOG") {
+        Ok(p) => std::fs::File::create(p).unwrap(),
+        Err(_) => std::fs::File::create("/tmp/darkfi.log").unwrap(),
+    };
+
+    simplelog::CombinedLogger::init(vec![
+        simplelog::TermLogger::new(
+            log_level,
+            log_config.clone(),
+            simplelog::TerminalMode::Mixed,
+            simplelog::ColorChoice::Auto,
+        ),
+        simplelog::WriteLogger::new(log_level, log_config, env_log_file_path),
+    ])?;
+
+    let url = Url::parse("tcp://127.0.0.1:55555").unwrap();
+
+    let settings = Settings {
+        inbound: Some(url),
+        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: None,
+        peers: Vec::new(),
+        seeds: Vec::new(),
+        node_id: String::new(),
+    };
+
+    let ex = Arc::new(Executor::new());
+    let ex2 = ex.clone();
+
+    let p2p = net::P2p::new(settings).await;
+
+    let seed = DchatSeed::new(p2p);
+
+    let nthreads = num_cpus::get();
+    let (signal, shutdown) = async_channel::unbounded::<()>();
+    let (_, result) = Parallel::new()
+        .each(0..nthreads, |_| smol::future::block_on(ex.run(shutdown.recv())))
+        .finish(|| {
+            smol::future::block_on(async move {
+                seed.start(ex2).await?;
+                drop(signal);
+                Ok(())
+            })
+        });
+
+    result
+}
+
+struct DchatSeed {
+    p2p: net::P2pPtr,
+}
+
+impl DchatSeed {
+    fn new(p2p: net::P2pPtr) -> Self {
+        Self { p2p }
+    }
+
+    async fn start(&self, executor: Arc<Executor<'_>>) -> Result<()> {
+        self.p2p.clone().start(executor.clone()).await?;
+
+        self.p2p.clone().run(executor.clone()).await?;
+
+        Ok(())
+    }
+}

example/dchat/Cargo.toml

@@ -0,0 +1,36 @@
+[package]
+name = "dchat"
+version = "0.1.0"
+edition = "2021"
+description = "Demo chat to document darkfi net code"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+darkfi = {path = "../../", features = ["net", "rpc"]}
+
+# Async
+futures = "0.3.0"
+async-std = "1"
+async-trait = "0.1.56"
+async-executor = "1.4.1"
+async-channel = "1.6.1"
+easy-parallel = "3.2.0"
+smol = "1.2.5"
+num_cpus = "1.13.1"
+
+# Misc
+log = "0.4.17"
+simplelog = "0.12.0"
+ringbuffer = "0.8.4"
+url = "2.2.2"
+clap = {version = "3.2.8", features = ["derive"]}
+termion = "1.5.6"
+
+# Encoding and parsing
+serde = {version = "1.0.138", features = ["derive"]}
+structopt = "0.3.26"
+structopt-toml = "0.5.0"
+toml = "0.4.2"
+
+

example/dchat/dchat_config.toml

@@ -0,0 +1,28 @@
+# chat toml
+
+[net]
+## P2P accept address
+#inbound="tls://127.0.0.1:11002" 
+
+## Connection slots
+outbound_connections=5
+
+## P2P external address
+#external_addr="tls://127.0.0.1:11002"
+
+## Peers to connect to
+#peers=["tls://127.0.0.1:11003"]
+
+## Seed nodes to connect to 
+#seeds=["tls://irc0.dark.fi:11001", "tls://irc1.dark.fi:11001"]
+seeds=["tcp://127.0.0.1:55555"]
+
+## Only used for debugging. Compromises privacy when set.
+#node_id = "foo"
+
+## these are the default configuration for the p2p network
+#manual_attempt_limit=0
+#seed_query_timeout_seconds=8
+#connect_timeout_seconds=10
+#channel_handshake_seconds=4
+#channel_heartbeat_seconds=10

example/dchat/src/dchatmsg.rs

@@ -0,0 +1,19 @@
+use async_std::sync::{Arc, Mutex};
+
+use darkfi::{
+    net,
+    util::serial::{SerialDecodable, SerialEncodable},
+};
+
+pub type DchatmsgsBuffer = Arc<Mutex<Vec<Dchatmsg>>>;
+
+impl net::Message for Dchatmsg {
+    fn name() -> &'static str {
+        "Dchatmsg"
+    }
+}
+
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct Dchatmsg {
+    pub message: String,
+}

example/dchat/src/error.rs

@@ -0,0 +1,15 @@
+use std::{error, fmt};
+
+pub type Error = Box<dyn error::Error>;
+pub type Result<T> = std::result::Result<T, Error>;
+
+#[derive(Debug, Clone)]
+pub struct MissingSpecifier;
+
+impl fmt::Display for MissingSpecifier {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "missing node specifier. you must specify either a or b")
+    }
+}
+
+impl error::Error for MissingSpecifier {}

example/dchat/src/main.rs

@@ -0,0 +1,337 @@
+use async_executor::Executor;
+use async_std::sync::{Arc, Mutex};
+use easy_parallel::Parallel;
+
+use std::{
+    fs::File,
+    io::{stdin, stdout, Read, Write},
+};
+
+use log::debug;
+use simplelog::WriteLogger;
+use url::Url;
+
+use termion::{event::Key, input::TermRead, raw::IntoRawMode};
+
+use darkfi::{
+    net,
+    net::Settings,
+    util::cli::{get_log_config, get_log_level},
+};
+
+use crate::{
+    dchatmsg::{Dchatmsg, DchatmsgsBuffer},
+    error::{Error, MissingSpecifier, Result},
+    protocol_dchat::ProtocolDchat,
+};
+
+pub mod dchatmsg;
+pub mod error;
+pub mod protocol_dchat;
+
+struct Dchat {
+    p2p: net::P2pPtr,
+    recv_msgs: DchatmsgsBuffer,
+    input: String,
+    display: DisplayMode,
+}
+
+enum DisplayMode {
+    Normal,
+    Editing,
+    Inbox,
+    MessageSent,
+    SendFailed(Error),
+}
+
+impl Dchat {
+    fn new(
+        p2p: net::P2pPtr,
+        recv_msgs: DchatmsgsBuffer,
+        input: String,
+        display: DisplayMode,
+    ) -> Self {
+        Self { p2p, recv_msgs, input, display }
+    }
+
+    async fn menu(&mut self) -> Result<()> {
+        debug!(target: "dchat", "Dchat::menu() [START]");
+        let stdout = stdout();
+        let mut stdout = stdout.lock().into_raw_mode().unwrap();
+        let mut stdin = stdin();
+
+        loop {
+            self.render().await?;
+            for k in stdin.by_ref().keys() {
+                match &self.display {
+                    DisplayMode::Normal => match k.unwrap() {
+                        Key::Char('q') => return Ok(()),
+                        Key::Char('i') => {
+                            self.display = DisplayMode::Inbox;
+                            break
+                        }
+
+                        Key::Char('s') => {
+                            self.display = DisplayMode::Editing;
+                            break
+                        }
+                        _ => {}
+                    },
+                    DisplayMode::Editing => match k.unwrap() {
+                        Key::Char('q') => return Ok(()),
+                        Key::Char('\n') => {
+                            match self.send().await {
+                                Ok(_) => {
+                                    self.display = DisplayMode::MessageSent;
+                                }
+                                Err(e) => {
+                                    self.display = DisplayMode::SendFailed(e);
+                                }
+                            }
+                            break
+                        }
+                        Key::Char(c) => {
+                            self.input.push(c);
+                        }
+                        Key::Esc => {
+                            self.display = DisplayMode::Normal;
+                            break
+                        }
+                        _ => {}
+                    },
+                    DisplayMode::MessageSent => match k.unwrap() {
+                        Key::Char('q') => return Ok(()),
+                        Key::Esc => {
+                            self.display = DisplayMode::Normal;
+                            break
+                        }
+                        _ => {}
+                    },
+                    DisplayMode::Inbox => match k.unwrap() {
+                        Key::Char('q') => return Ok(()),
+                        _ => {}
+                    },
+                    DisplayMode::SendFailed(_) => match k.unwrap() {
+                        Key::Char('q') => return Ok(()),
+                        Key::Esc => {
+                            self.display = DisplayMode::Normal;
+                            break
+                        }
+                        _ => {}
+                    },
+                }
+            }
+            stdout.flush()?;
+        }
+    }
+
+    async fn render(&mut self) -> Result<()> {
+        debug!(target: "dchat", "Dchat::render() [START]");
+        let stdout = stdout();
+        let mut stdout = stdout.lock().into_raw_mode().unwrap();
+
+        match &self.display {
+            DisplayMode::Normal => {
+                write!(
+                    stdout,
+                    "{}{}{}Welcome to dchat. {} s: send message {} i: inbox {} q: quit {}",
+                    termion::clear::All,
+                    termion::style::Bold,
+                    termion::cursor::Goto(1, 2),
+                    termion::cursor::Goto(1, 3),
+                    termion::cursor::Goto(1, 4),
+                    termion::cursor::Goto(1, 5),
+                    termion::cursor::Goto(1, 6)
+                )?;
+                stdout.flush()?;
+            }
+            DisplayMode::Editing => {
+                write!(
+                    stdout,
+                    "{}{}{}enter your msg.{} esc: stop editing {} enter: send {}",
+                    termion::clear::All,
+                    termion::style::Bold,
+                    termion::cursor::Goto(1, 2),
+                    termion::cursor::Goto(1, 3),
+                    termion::cursor::Goto(1, 4),
+                    termion::cursor::Goto(1, 5)
+                )?;
+                stdout.flush()?;
+            }
+            DisplayMode::Inbox => {
+                let msgs = self.recv_msgs.lock().await;
+                for i in msgs.iter() {
+                    if !i.message.is_empty() {
+                        write!(
+                            stdout,
+                            "{}{}{}received msg: {}",
+                            termion::clear::All,
+                            termion::style::Bold,
+                            termion::cursor::Goto(1, 2),
+                            i.message
+                        )?;
+                    } else {
+                        write!(
+                            stdout,
+                            "{}{}{}inbox is empty",
+                            termion::clear::All,
+                            termion::style::Bold,
+                            termion::cursor::Goto(1, 2),
+                        )?;
+                    }
+                }
+                stdout.flush()?;
+            }
+            DisplayMode::MessageSent => {
+                write!(
+                    stdout,
+                    "{}{}{}message sent! {} esc: return to main menu {}",
+                    termion::clear::All,
+                    termion::style::Bold,
+                    termion::cursor::Goto(1, 2),
+                    termion::cursor::Goto(1, 3),
+                    termion::cursor::Goto(1, 4),
+                )?;
+                stdout.flush()?;
+            }
+            DisplayMode::SendFailed(e) => {
+                write!(
+                    stdout,
+                    "{}{}{}send message failed! reason: {} {} esc: return to main menu {}",
+                    termion::clear::All,
+                    termion::style::Bold,
+                    termion::cursor::Goto(1, 2),
+                    e,
+                    termion::cursor::Goto(1, 3),
+                    termion::cursor::Goto(1, 4),
+                )?;
+                stdout.flush()?;
+            }
+        }
+
+        Ok(())
+    }
+    async fn register_protocol(&self, msgs: DchatmsgsBuffer) -> Result<()> {
+        debug!(target: "dchat", "Dchat::register_protocol() [START]");
+        let registry = self.p2p.protocol_registry();
+        registry
+            .register(net::SESSION_ALL, move |channel, _p2p| {
+                let msgs2 = msgs.clone();
+                async move { ProtocolDchat::init(channel, msgs2).await }
+            })
+            .await;
+        debug!(target: "dchat", "Dchat::register_protocol() [STOP]");
+        Ok(())
+    }
+
+    async fn start(&self, ex: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "dchat", "Dchat::start() [START]");
+
+        let ex2 = ex.clone();
+
+        self.register_protocol(self.recv_msgs.clone()).await?;
+        self.p2p.clone().start(ex.clone()).await?;
+        ex2.spawn(self.p2p.clone().run(ex.clone())).detach();
+
+        debug!(target: "dchat", "Dchat::start() [STOP]");
+        Ok(())
+    }
+
+    async fn send(&self) -> Result<()> {
+        let message = self.input.clone();
+        let dchatmsg = Dchatmsg { message };
+        self.p2p.broadcast(dchatmsg).await?;
+        Ok(())
+    }
+}
+
+// inbound
+fn alice() -> Result<Settings> {
+    let seed = Url::parse("tcp://127.0.0.1:55555").unwrap();
+    let inbound = Url::parse("tcp://127.0.0.1:55554").unwrap();
+    let ext_addr = Url::parse("tcp://127.0.0.1:55554").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(),
+    };
+
+    Ok(settings)
+}
+
+// outbound
+fn bob() -> Result<Settings> {
+    let seed = Url::parse("tcp://127.0.0.1:55555").unwrap();
+    let oc = 5;
+
+    let settings = Settings {
+        inbound: None,
+        outbound_connections: oc,
+        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: None,
+        peers: Vec::new(),
+        seeds: vec![seed],
+        node_id: String::new(),
+    };
+
+    Ok(settings)
+}
+
+#[async_std::main]
+async fn main() -> Result<()> {
+    let log_level = get_log_level(1);
+    let log_config = get_log_config();
+
+    let log_path = "/tmp/dchat.log";
+    let file = File::create(log_path).unwrap();
+    WriteLogger::init(log_level, log_config, file)?;
+
+    let settings: Result<Settings> = match std::env::args().nth(1) {
+        Some(id) => match id.as_str() {
+            "a" => alice(),
+            "b" => bob(),
+            _ => Err(MissingSpecifier.into()),
+        },
+        None => Err(MissingSpecifier.into()),
+    };
+
+    let p2p = net::P2p::new(settings?.into()).await;
+
+    let nthreads = num_cpus::get();
+    let (signal, shutdown) = async_channel::unbounded::<()>();
+
+    let ex = Arc::new(Executor::new());
+    let ex2 = ex.clone();
+
+    let msgs: DchatmsgsBuffer = Arc::new(Mutex::new(vec![Dchatmsg { message: String::new() }]));
+
+    let mut dchat = Dchat::new(p2p, msgs, String::new(), DisplayMode::Normal);
+
+    let (_, result) = Parallel::new()
+        .each(0..nthreads, |_| smol::future::block_on(ex.run(shutdown.recv())))
+        .finish(|| {
+            smol::future::block_on(async move {
+                dchat.start(ex2).await?;
+                dchat.menu().await?;
+                drop(signal);
+                Ok(())
+            })
+        });
+
+    result
+}

example/dchat/src/protocol_dchat.rs

@@ -0,0 +1,55 @@
+use async_executor::Executor;
+use async_std::sync::Arc;
+use async_trait::async_trait;
+use darkfi::{net, Result};
+use log::debug;
+
+use crate::dchatmsg::{Dchatmsg, DchatmsgsBuffer};
+
+pub struct ProtocolDchat {
+    jobsman: net::ProtocolJobsManagerPtr,
+    msg_sub: net::MessageSubscription<Dchatmsg>,
+    msgs: DchatmsgsBuffer,
+}
+
+impl ProtocolDchat {
+    pub async fn init(channel: net::ChannelPtr, msgs: DchatmsgsBuffer) -> net::ProtocolBasePtr {
+        debug!(target: "dchat", "ProtocolDchat::init() [START]");
+        let message_subsytem = channel.get_message_subsystem();
+        message_subsytem.add_dispatch::<Dchatmsg>().await;
+
+        let msg_sub =
+            channel.subscribe_msg::<Dchatmsg>().await.expect("Missing DchatMsg dispatcher!");
+
+        Arc::new(Self {
+            jobsman: net::ProtocolJobsManager::new("ProtocolDchat", channel.clone()),
+            msg_sub,
+            msgs,
+        })
+    }
+
+    async fn handle_receive_msg(self: Arc<Self>) -> Result<()> {
+        debug!(target: "dchat", "ProtocolDchat::handle_receive_msg() [START]");
+        while let Ok(msg) = self.msg_sub.receive().await {
+            let msg = (*msg).to_owned();
+            self.msgs.lock().await.push(msg);
+        }
+
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl net::ProtocolBase for ProtocolDchat {
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "dchat", "ProtocolDchat::ProtocolBase::start() [START]");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().handle_receive_msg(), executor.clone()).await;
+        debug!(target: "dchat", "ProtocolDchat::ProtocolBase::start() [STOP]");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolDchat"
+    }
+}