From 70fb6ebd3c6c3500b2b105940e3928c2abfb1040 Mon Sep 17 00:00:00 2001 From: Daniel Sanchez Quiros Date: Tue, 18 Oct 2022 19:48:24 +0200 Subject: [PATCH] Implement chat example main block --- examples/toy-chat/Cargo.toml | 3 +- examples/toy-chat/src/main.rs | 146 +++++++++++++++++++++++++++--- examples/toy-chat/src/protocol.rs | 27 ++++-- waku/src/lib.rs | 4 +- waku/src/node/mod.rs | 10 +- waku/src/node/peers.rs | 2 +- waku/src/node/store.rs | 16 ++-- 7 files changed, 171 insertions(+), 37 deletions(-) diff --git a/examples/toy-chat/Cargo.toml b/examples/toy-chat/Cargo.toml index 5108f94..99ed72f 100644 --- a/examples/toy-chat/Cargo.toml +++ b/examples/toy-chat/Cargo.toml @@ -11,4 +11,5 @@ tui = "0.19" crossterm = "0.25" unicode-width = "0.1" prost = "0.11" -once_cell = "1.15" \ No newline at end of file +once_cell = "1.15" +chrono = "0.4" \ No newline at end of file diff --git a/examples/toy-chat/src/main.rs b/examples/toy-chat/src/main.rs index 682fbca..d0e27ed 100644 --- a/examples/toy-chat/src/main.rs +++ b/examples/toy-chat/src/main.rs @@ -1,10 +1,17 @@ mod protocol; +use crate::protocol::{Chat2Message, TOY_CHAT_CONTENT_TOPIC}; +use chrono::Utc; use crossterm::{ event::{self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode}, execute, terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, }; +use prost::Message; +use std::io::Write; +use std::str::FromStr; +use std::sync::{Arc, RwLock}; +use std::time::Duration; use std::{error::Error, io}; use tui::{ backend::{Backend, CrosstermBackend}, @@ -15,34 +22,93 @@ use tui::{ Frame, Terminal, }; use unicode_width::UnicodeWidthStr; -use waku::{waku_new, Result, WakuNodeConfig, WakuNodeHandle}; +use waku::{ + waku_new, waku_set_event_callback, ContentFilter, Multiaddr, ProtocolId, Running, StoreQuery, + WakuMessage, WakuNodeHandle, +}; enum InputMode { Normal, Editing, } +const NODES: &[&str] = &[ + "/dns4/node-01.ac-cn-hongkong-c.wakuv2.test.statusim.net/tcp/30303/p2p/16Uiu2HAkvWiyFsgRhuJEb9JfjYxEkoHLgnUQmr1N5mKWnYjxYRVm", + "/dns4/node-01.do-ams3.wakuv2.test.statusim.net/tcp/30303/p2p/16Uiu2HAmPLe7Mzm8TsYUubgCAW1aJoeFScxrLj8ppHFivPo97bUZ", + "/dns4/node-01.gc-us-central1-a.wakuv2.test.statusim.net/tcp/30303/p2p/16Uiu2HAmJb2e28qLXxT5kZxVUUoJt72EMzNGXB47Rxx5hw3q4YjS" +]; + /// App holds the state of the application struct App { /// Current value of the input box input: String, + nick: String, /// Current input mode input_mode: InputMode, /// History of recorded messages - messages: Vec, + messages: Arc>>, + + node_handle: WakuNodeHandle, } -impl Default for App { - fn default() -> App { +impl App { + fn new(nick: String, node_handle: WakuNodeHandle) -> App { App { input: String::new(), input_mode: InputMode::Normal, - messages: Vec::new(), + messages: Arc::new(RwLock::new(Vec::new())), + node_handle, + nick, } } } +fn retrieve_history(node_handle: &WakuNodeHandle) -> waku::Result> { + let self_id = node_handle.peer_id().unwrap(); + let peer = node_handle + .peers()? + .iter() + .cloned() + .find(|peer| peer.peer_id() != &self_id) + .unwrap(); + + let result = node_handle.store_query( + &StoreQuery { + pubsub_topic: None, + content_filters: vec![ContentFilter::new(TOY_CHAT_CONTENT_TOPIC.clone())], + start_time: Some( + (Duration::from_secs(Utc::now().timestamp() as u64) + - Duration::from_secs(60 * 60 * 24)) + .as_secs() as usize, + ), + end_time: None, + paging_options: None, + }, + peer.peer_id(), + Some(Duration::from_secs(10)), + )?; + + Ok(result + .messages() + .iter() + .map(|waku_message| { + ::decode(waku_message.payload()) + .expect("Toy chat messages should be decodeable") + }) + .collect()) +} +fn setup_node_handle() -> std::result::Result, Box> { + let node_handle = waku_new(None)?; + let node_handle = node_handle.start()?; + for address in NODES.iter().map(|a| Multiaddr::from_str(a).unwrap()) { + let peerid = node_handle.add_peer(&address, ProtocolId::Relay)?; + node_handle.connect_peer_with_id(peerid, None)?; + } + node_handle.relay_subscribe(None)?; + Ok(node_handle) +} fn main() -> std::result::Result<(), Box> { + let nick = std::env::args().nth(1).expect("Nick to be set"); // setup terminal enable_raw_mode()?; let mut stdout = io::stdout(); @@ -50,9 +116,34 @@ fn main() -> std::result::Result<(), Box> { let backend = CrosstermBackend::new(stdout); let mut terminal = Terminal::new(backend)?; + let node_handle = setup_node_handle()?; + // create app and run it - let app = App::default(); - let res = run_app(&mut terminal, app); + let mut app = App::new(nick, node_handle); + let history = retrieve_history(&app.node_handle)?; + if !history.is_empty() { + *app.messages.write().unwrap() = history; + } + let shared_messages = Arc::clone(&app.messages); + waku_set_event_callback(move |signal| match signal.event() { + waku::Event::WakuMessage(event) => { + match ::decode(event.waku_message().payload()) { + Ok(chat_message) => { + shared_messages.write().unwrap().push(chat_message); + } + Err(e) => { + let mut out = std::io::stderr(); + write!(out, "{:?}", e).unwrap(); + } + } + } + _ => { + unreachable!() + } + }); + + // app.node_handle.relay_publish_message(&WakuMessage::new(Chat2Message::new(&app.nick, format!("")))) + let res = run_app(&mut terminal, &mut app); // restore terminal disable_raw_mode()?; @@ -62,17 +153,20 @@ fn main() -> std::result::Result<(), Box> { DisableMouseCapture )?; terminal.show_cursor()?; + app.node_handle.stop()?; if let Err(err) = res { println!("{:?}", err) } - Ok(()) } -fn run_app(terminal: &mut Terminal, mut app: App) -> io::Result<()> { +fn run_app( + terminal: &mut Terminal, + app: &mut App, +) -> std::result::Result<(), Box> { loop { - terminal.draw(|f| ui(f, &app))?; + terminal.draw(|f| ui(f, app))?; if let Event::Key(key) = event::read()? { match app.input_mode { @@ -87,7 +181,23 @@ fn run_app(terminal: &mut Terminal, mut app: App) -> io::Result<( }, InputMode::Editing => match key.code { KeyCode::Enter => { - app.messages.push(app.input.drain(..).collect()); + let message_content: String = app.input.drain(..).collect(); + let message = Chat2Message::new(&app.nick, &message_content); + let mut buff = Vec::new(); + Message::encode(&message, &mut buff)?; + let waku_message = WakuMessage::new( + buff, + TOY_CHAT_CONTENT_TOPIC.clone(), + 1, + Utc::now().timestamp() as usize, + ); + if let Err(e) = + app.node_handle + .relay_publish_message(&waku_message, None, None) + { + let mut out = std::io::stderr(); + write!(out, "{:?}", e).unwrap(); + } } KeyCode::Char(c) => { app.input.push(c); @@ -126,7 +236,7 @@ fn ui(f: &mut Frame, app: &App) { Span::styled("q", Style::default().add_modifier(Modifier::BOLD)), Span::raw(" to exit, "), Span::styled("e", Style::default().add_modifier(Modifier::BOLD)), - Span::raw(" to start editing."), + Span::raw(" to start writing a message."), ], Style::default().add_modifier(Modifier::RAPID_BLINK), ), @@ -171,10 +281,16 @@ fn ui(f: &mut Frame, app: &App) { let messages: Vec = app .messages + .read() + .unwrap() .iter() - .enumerate() - .map(|(i, m)| { - let content = vec![Spans::from(Span::raw(format!("{}: {}", i, m)))]; + .map(|message| { + let content = vec![Spans::from(Span::raw(format!( + "[{} - {}]: {}", + message.timestamp().format("%d-%m-%y %H:%M"), + message.nick(), + message.message() + )))]; ListItem::new(content) }) .collect(); diff --git a/examples/toy-chat/src/protocol.rs b/examples/toy-chat/src/protocol.rs index 0181204..110747e 100644 --- a/examples/toy-chat/src/protocol.rs +++ b/examples/toy-chat/src/protocol.rs @@ -1,11 +1,9 @@ -use once_cell::sync::{Lazy, OnceCell}; -use prost::{ - encoding::{bytes, string, uint64}, - Message, -}; -use waku::{Encoding, WakuContentTopic, WakuMessage}; +use chrono::{DateTime, TimeZone, Utc}; +use once_cell::sync::Lazy; +use prost::Message; +use waku::{Encoding, WakuContentTopic}; -const TOY_CHAT_CONTENT_TOPIC: Lazy = Lazy::new(|| WakuContentTopic { +pub static TOY_CHAT_CONTENT_TOPIC: Lazy = Lazy::new(|| WakuContentTopic { application_name: "toy-chat".into(), version: 2, content_topic_name: "huilong".into(), @@ -23,7 +21,22 @@ pub struct Chat2Message { } impl Chat2Message { + pub fn new(nick: &str, payload: &str) -> Self { + Self { + timestamp: Utc::now().timestamp() as u64, + nick: nick.to_string(), + payload: payload.as_bytes().to_vec(), + } + } pub fn message(&self) -> String { String::from_utf8(self.payload.clone()).unwrap() } + + pub fn nick(&self) -> &str { + &self.nick + } + + pub fn timestamp(&self) -> DateTime { + Utc.timestamp(self.timestamp as i64, 0) + } } diff --git a/waku/src/lib.rs b/waku/src/lib.rs index cc7f979..109cdb3 100644 --- a/waku/src/lib.rs +++ b/waku/src/lib.rs @@ -8,8 +8,8 @@ mod node; pub use node::{ waku_create_content_topic, waku_create_pubsub_topic, waku_dafault_pubsub_topic, waku_new, - waku_store_query, Initialized, Protocol, Running, WakuNodeConfig, WakuNodeHandle, WakuPeerData, - WakuPeers, + waku_store_query, Aes256Gcm, Initialized, Key, Multiaddr, Protocol, PublicKey, Running, + SecretKey, WakuNodeConfig, WakuNodeHandle, WakuPeerData, WakuPeers, }; pub use general::{ diff --git a/waku/src/node/mod.rs b/waku/src/node/mod.rs index 9ad554b..e98eff9 100644 --- a/waku/src/node/mod.rs +++ b/waku/src/node/mod.rs @@ -9,9 +9,9 @@ mod relay; mod store; // std -use aes_gcm::{Aes256Gcm, Key}; -use multiaddr::Multiaddr; -use secp256k1::{PublicKey, SecretKey}; +pub use aes_gcm::{Aes256Gcm, Key}; +pub use multiaddr::Multiaddr; +pub use secp256k1::{PublicKey, SecretKey}; use std::marker::PhantomData; use std::sync::Mutex; use std::time::Duration; @@ -232,8 +232,8 @@ impl WakuNodeHandle { pub fn store_query( &self, query: &StoreQuery, - peer_id: PeerId, - timeout: Duration, + peer_id: &PeerId, + timeout: Option, ) -> Result { store::waku_store_query(query, peer_id, timeout) } diff --git a/waku/src/node/peers.rs b/waku/src/node/peers.rs index dea603f..dcd7abb 100644 --- a/waku/src/node/peers.rs +++ b/waku/src/node/peers.rs @@ -125,7 +125,7 @@ pub fn waku_peer_count() -> Result { pub type Protocol = String; /// Peer data from known/connected waku nodes -#[derive(Deserialize)] +#[derive(Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct WakuPeerData { /// Waku peer id diff --git a/waku/src/node/store.rs b/waku/src/node/store.rs index cc39a7d..818b10e 100644 --- a/waku/src/node/store.rs +++ b/waku/src/node/store.rs @@ -13,8 +13,8 @@ use crate::general::{JsonResponse, PeerId, Result, StoreQuery, StoreResponse}; /// These [`PagingOptions`](`crate::general::PagingOptions`) must contain a cursor pointing to the Index from which a new page can be requested pub fn waku_store_query( query: &StoreQuery, - peer_id: PeerId, - timeout: Duration, + peer_id: &PeerId, + timeout: Option, ) -> Result { let result = unsafe { CStr::from_ptr(waku_sys::waku_store_query( @@ -24,13 +24,17 @@ pub fn waku_store_query( ) .expect("CString should build properly from the serialized filter subscription") .into_raw(), - CString::new(peer_id) + CString::new(peer_id.clone()) .expect("CString should build properly from peer id") .into_raw(), timeout - .as_millis() - .try_into() - .expect("Duration as milliseconds should fit in a i32"), + .map(|timeout| { + timeout + .as_millis() + .try_into() + .expect("Duration as milliseconds should fit in a i32") + }) + .unwrap_or(0), )) } .to_str()