dchat: basic cli that sends messages over tpc.

TODO: connect to p2p

example/dchat/Cargo.toml

@@ -0,0 +1,32 @@
+[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"
+
+# Misc
+simplelog = "0.12.0"
+ringbuffer = "0.8.4"
+url = "2.2.2"
+clap = {version = "3.2.8", features = ["derive"]}
+
+# Encoding and parsing
+serde = {version = "1.0.138", features = ["derive"]}
+structopt = "0.3.26"
+structopt-toml = "0.5.0"
+
+

example/dchat/chat_config.toml

@@ -0,0 +1,27 @@
+# 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"]
+
+## 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},
+};
+use ringbuffer::AllocRingBuffer;
+
+impl net::Message for Dchatmsg {
+    fn name() -> &'static str {
+        "Dchatmsg"
+    }
+}
+
+pub type DchatmsgsBuffer = Arc<Mutex<AllocRingBuffer<Dchatmsg>>>;
+
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct Dchatmsg {
+    pub message: String,
+}

example/dchat/src/main.rs

@@ -0,0 +1,208 @@
+use async_channel::Receiver;
+use async_executor::Executor;
+use async_std::{
+    net::{TcpListener, TcpStream},
+    sync::{Arc, Mutex},
+};
+use clap::{Parser, Subcommand};
+use futures::{io::WriteHalf, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use smol::Async;
+use url::Url;
+
+use darkfi::{
+    async_daemonize, cli_desc, net,
+    system::{Subscriber, SubscriberPtr},
+    util::{
+        cli::{get_log_config, get_log_level, spawn_config},
+        path::get_config_path,
+    },
+    Result,
+};
+use simplelog::{ColorChoice, TermLogger, TerminalMode};
+use smol::future;
+use structopt_toml::StructOptToml;
+
+use crate::{
+    dchatmsg::{Dchatmsg, DchatmsgsBuffer},
+    protocol_dchat::ProtocolDchat,
+    settings::{CONFIG_FILE, CONFIG_FILE_CONTENTS},
+};
+
+pub mod dchatmsg;
+pub mod protocol_dchat;
+pub mod server;
+pub mod settings;
+
+const SIZE_OF_MSGS_BUFFER: usize = 4096;
+
+#[derive(Parser)]
+#[clap(name = "dchat", about = cli_desc!(), version)]
+#[clap(arg_required_else_help(true))]
+struct Args {
+    #[clap(short, parse(from_occurrences))]
+    /// Increase verbosity (-vvv supported)
+    verbose: u8,
+
+    #[clap(subcommand)]
+    command: Option<Dchatsubcommand>,
+}
+
+#[derive(Subcommand)]
+enum Dchatsubcommand {
+    Inbox,
+    Send { msg: String, addr: String },
+    Receive { addr: String },
+}
+
+#[async_std::main]
+async fn main() -> Result<()> {
+    let args = Args::parse();
+
+    let log_level = get_log_level(args.verbose.into());
+    let log_config = get_log_config();
+    TermLogger::init(log_level, log_config, TerminalMode::Mixed, ColorChoice::Auto)?;
+
+    let dchat = Dchat::new();
+
+    match args.command {
+        Some(sc) => match sc {
+            Dchatsubcommand::Inbox => {
+                eprintln!("inbox");
+            }
+            Dchatsubcommand::Send { msg, addr } => {
+                dchat.send(msg, addr).await?;
+            }
+            Dchatsubcommand::Receive { addr } => {
+                dchat.receive(addr).await?;
+            }
+        },
+        None => {}
+    }
+    Ok(())
+}
+struct Dchat {
+    //dchatmsgs_buffer: DchatmsgsBuffer,
+    //p2p: net::P2pPtr,
+    //senders: SubscriberPtr<Dchatmsg>,
+}
+
+impl Dchat {
+    //fn new(dchatmsgs_buffer: DchatmsgsBuffer, p2p: net::P2pPtr) -> Self {
+    //    let senders = Subscriber::new();
+    //    Self { dchatmsgs_buffer, p2p, senders }
+    //}
+    fn new() -> Arc<Self> {
+        Arc::new(Self {})
+    }
+
+    async fn receive(self: Arc<Self>, addr: String) -> Result<()> {
+        smol::block_on(async {
+            let listener = TcpListener::bind(&addr).await?;
+            eprintln!("Listening on {}", listener.local_addr()?);
+            loop {
+                let (stream, peer_addr) = listener.accept().await?;
+                println!("Accepted client: {}", peer_addr);
+                smol::spawn(self.clone().read_msg(stream)).detach();
+            }
+        })
+    }
+
+    async fn send(&self, msg: String, addr: String) -> Result<()> {
+        let mut stream = TcpStream::connect(&addr).await?;
+        eprintln!("Connected to {}", stream.local_addr()?);
+        stream.write_all(msg.as_bytes()).await?;
+        eprintln!("Sending '{}'", msg);
+        Ok(())
+    }
+
+    async fn read_msg(self: Arc<Self>, mut stream: TcpStream) -> Result<()> {
+        let mut buffer = [0u8; 4];
+        stream.read_exact(&mut buffer).await?;
+        let buffer = std::str::from_utf8(&buffer).unwrap();
+        eprintln!("{}", buffer);
+        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 {
+    //                senders.notify(msg).await;
+    //            }
+    //        })
+    //        .detach();
+    //}
+}
+
+//#[async_std::main]
+//async_daemonize!(realmain);
+//async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
+//    //let dchatmsgs_buffer: DchatmsgsBuffer =
+//    //    Arc::new(Mutex::new(ringbuffer::AllocRingBuffer::with_capacity(SIZE_OF_MSGS_BUFFER)));
+//    //// Pick up channel settings from the TOML configuration
+//    //let cfg_path = get_config_path(settings.config, CONFIG_FILE)?;
+//
+//    //let net_settings = settings.net;
+//    //let (p2p_send_channel, p2p_recv_channel) = async_channel::unbounded::<Dchatmsg>();
+//    //let p2p = net::P2p::new(net_settings.into()).await;
+//    //let p2p = p2p.clone();
+//
+//    //let registry = p2p.protocol_registry();
+//
+//    //let dchatmsgs_buffer_cloned = dchatmsgs_buffer.clone();
+//
+//    //registry
+//    //    .register(net::SESSION_ALL, move |channel, p2p| {
+//    //        let sender = p2p_send_channel.clone();
+//    //        let privmsgs_buffer_cloned = dchatmsgs_buffer_cloned.clone();
+//    //        async move { ProtocolDchat::init(channel, sender, p2p, privmsgs_buffer_cloned).await }
+//    //    })
+//    //    .await;
+//
+//    //p2p.clone().start(executor.clone()).await?;
+//
+//    //let executor_cloned = executor.clone();
+//    //executor_cloned.spawn(p2p.clone().run(executor.clone())).detach();
+//
+//    //let listenaddr = settings.listen.socket_addrs(|| None)?[0];
+//    //let listener = TcpListener::bind(listenaddr).await?;
+//
+//    //let executor_cloned = executor.clone();
+//    //executor
+//    //    .spawn(async move {
+//    //        let dchat = Dchatd::new(dchatmsgs_buffer.clone(), p2p.clone());
+//
+//    //        dchat.start_p2p_receive_loop(executor_cloned.clone(), p2p_recv_channel);
+//
+//    //        loop {
+//    //            let (stream, peer_addr) = match listener.accept().await {
+//    //                Ok((s, a)) => (s, a),
+//    //                Err(e) => {
+//    //                    //error!("failed accepting new connections: {}", e);
+//    //                    continue;
+//    //                }
+//    //            };
+//
+//    //            //ircd.process_new_connection(executor_cloned.clone(), stream, peer_addr).await
+//
+//    //            //if let Err(e) = result {
+//    //            //    error!("Failed processing connection {}: {}", peer_addr, e);
+//    //            //    continue;
+//    //            //};
+//
+//    //            //info!("IRC Accepted new client: {}", peer_addr);
+//    //        }
+//    //    })
+//    //    .detach();
+//
+//    //let (signal, shutdown) = async_channel::bounded::<()>(1);
+//
+//    //
+//    //shutdown.recv().await?;
+//
+//    Ok(())
+//}

example/dchat/src/protocol_dchat.rs

@@ -0,0 +1,79 @@
+use async_executor::Executor;
+use async_std::sync::Arc;
+use async_trait::async_trait;
+use darkfi::{net, Result};
+
+use crate::dchatmsg::{Dchatmsg, DchatmsgsBuffer};
+use ringbuffer::{RingBufferExt, RingBufferWrite};
+
+pub struct ProtocolDchat {
+    jobsman: net::ProtocolJobsManagerPtr,
+    notify_queue_sender: async_channel::Sender<Dchatmsg>,
+    msg_sub: net::MessageSubscription<Dchatmsg>,
+    p2p: net::P2pPtr,
+    msgs: DchatmsgsBuffer,
+    channel: net::ChannelPtr,
+}
+
+impl ProtocolDchat {
+    pub async fn init(
+        channel: net::ChannelPtr,
+        notify_queue_sender: async_channel::Sender<Dchatmsg>,
+        p2p: net::P2pPtr,
+        msgs: DchatmsgsBuffer,
+    ) -> net::ProtocolBasePtr {
+        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 {
+            notify_queue_sender,
+            msg_sub,
+            jobsman: net::ProtocolJobsManager::new("ProtocolDchat", channel.clone()),
+            p2p,
+            msgs,
+            channel,
+        })
+    }
+
+    async fn handle_receive_msg(self: Arc<Self>) -> Result<()> {
+        let exclude_list = vec![self.channel.address()];
+
+        let msgs_buffer = self.msgs.lock().await;
+        let msgs = msgs_buffer.to_vec();
+        drop(msgs_buffer);
+        for m in msgs {
+            self.channel.send(m.clone()).await?;
+        }
+
+        loop {
+            let msg = self.msg_sub.receive().await?;
+            let mut msg = (*msg).to_owned();
+
+            // add the msg to the buffer
+            self.msgs.lock().await.push(msg.clone());
+
+            self.notify_queue_sender.send(msg.clone()).await?;
+
+            self.p2p.broadcast_with_exclude(msg, &exclude_list).await?;
+        }
+    }
+}
+
+#[async_trait]
+impl net::ProtocolBase for ProtocolDchat {
+    /// Starts ping-pong keep-alive messages exchange. Runs ping-pong in the
+    /// protocol task manager, then queues the reply. Sends out a ping and
+    /// waits for pong reply. Waits for ping and replies with a pong.
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().handle_receive_msg(), executor.clone()).await;
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolDchat"
+    }
+}

example/dchat/src/server.rs

@@ -0,0 +1,59 @@
+use futures::{io::WriteHalf, AsyncRead, AsyncWrite, AsyncWriteExt};
+use ringbuffer::{RingBufferExt, RingBufferWrite};
+use std::net::SocketAddr;
+
+use darkfi::{net::P2pPtr, system::SubscriberPtr, Result};
+
+use crate::dchatmsg::{Dchatmsg, DchatmsgsBuffer};
+
+pub struct DchatserverConnection<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> {
+    // server stream
+    write_stream: WriteHalf<C>,
+    peer_address: SocketAddr,
+    // msgs
+    dchatmsgs_buffer: DchatmsgsBuffer,
+    // p2p
+    p2p: P2pPtr,
+    senders: SubscriberPtr<Dchatmsg>,
+    subscriber_id: u64,
+}
+
+impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> DchatserverConnection<C> {
+    pub fn new(
+        write_stream: WriteHalf<C>,
+        peer_address: SocketAddr,
+        dchatmsgs_buffer: DchatmsgsBuffer,
+        p2p: P2pPtr,
+        senders: SubscriberPtr<Dchatmsg>,
+        subscriber_id: u64,
+    ) -> Self {
+        Self { write_stream, peer_address, dchatmsgs_buffer, p2p, senders, subscriber_id }
+    }
+
+    async fn reply(&mut self, message: &str) -> Result<()> {
+        self.write_stream.write_all(message.as_bytes()).await?;
+        //debug!("Sent {}", message);
+        Ok(())
+    }
+
+    async fn update(&mut self, line: String) -> Result<()> {
+        // read from STDIN??
+        let mut tokens = line.split_ascii_whitespace();
+        Ok(())
+    }
+
+    async fn on_receive_dchatmsg(&mut self, message: &str, target: &str) -> Result<()> {
+        let protocol_msg = Dchatmsg { message: message.to_string() };
+
+        {
+            (*self.dchatmsgs_buffer.lock().await).push(protocol_msg.clone())
+        }
+
+        self.senders.notify_with_exclude(protocol_msg.clone(), &[self.subscriber_id]).await;
+
+        //debug!(target: "ircd", "PRIVMSG to be sent: {:?}", protocol_msg);
+        self.p2p.broadcast(protocol_msg).await?;
+
+        Ok(())
+    }
+}

example/dchat/src/settings.rs

@@ -0,0 +1,28 @@
+use darkfi::net::settings::SettingsOpt;
+use serde::Deserialize;
+use structopt::StructOpt;
+use structopt_toml::StructOptToml;
+use url::Url;
+
+pub const CONFIG_FILE: &str = "chat_config.toml";
+pub const CONFIG_FILE_CONTENTS: &str = include_str!("../chat_config.toml");
+
+#[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
+#[serde(default)]
+#[structopt(name = "chatapp")]
+pub struct Args {
+    /// Sets a custom config file
+    #[structopt(long)]
+    pub config: Option<String>,
+
+    /// IRC listen URL
+    #[structopt(long = "listen", default_value = "tcp://127.0.0.1:11066")]
+    pub listen: Url,
+
+    #[structopt(flatten)]
+    pub net: SettingsOpt,
+
+    /// Increase verbosity
+    #[structopt(short, parse(from_occurrences))]
+    pub verbose: u8,
+}