From 3f72fc7a2d2f4851550645a0582563307c81a0e3 Mon Sep 17 00:00:00 2001 From: Ivan Folgueira Bande Date: Sun, 1 Dec 2024 20:58:33 +0100 Subject: [PATCH] progress in toy-chat --- examples/toy-chat/src/main.rs | 247 +++++++++++-------------------- waku-bindings/src/general/mod.rs | 38 ++++- waku-bindings/src/node/config.rs | 4 + waku-bindings/src/node/mod.rs | 1 - waku-bindings/src/node/store.rs | 6 +- 5 files changed, 129 insertions(+), 167 deletions(-) diff --git a/examples/toy-chat/src/main.rs b/examples/toy-chat/src/main.rs index 90fc3e3..1d0e200 100644 --- a/examples/toy-chat/src/main.rs +++ b/examples/toy-chat/src/main.rs @@ -8,10 +8,10 @@ use crossterm::{ terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, }; use prost::Message; -use std::str::from_utf8; use std::io::Write; use std::sync::{Arc, RwLock}; use std::{error::Error, io}; +use std::time::Duration; use tui::{ backend::{Backend, CrosstermBackend}, layout::{Constraint, Direction, Layout}, @@ -85,63 +85,30 @@ impl App { let shared_messages = Arc::clone(&self.messages); self.waku.set_event_callback(move|response| { - // if let LibwakuResponse::Success(v) = response { - // let event: WakuEvent = - // serde_json::from_str(v.unwrap().as_str()).expect("Parsing event to succeed"); - - // match event { - // WakuEvent::WakuMessage(evt) => { - // // println!("WakuMessage event received: {:?}", evt.waku_message); - // let message = evt.waku_message; - // let payload = message.payload.to_vec(); - // match from_utf8(&payload) { - // Ok(msg) => { - // // Lock succeeded, proceed to send the message - // // if tx_clone.blocking_send(msg.to_string()).is_err() { - // // eprintln!("Failed to send message to async task"); - // // } - // } - // Err(e) => { - // eprintln!("Failed to decode payload as UTF-8: {}", e); - // // Handle the error as needed, or just log and skip - // } - // } - // } - // WakuEvent::Unrecognized(err) => eprintln!("Unrecognized waku event: {:?}", err), - // _ => eprintln!("event case not expected"), - // }; - // } - - if let LibwakuResponse::Success(v) = response { - if let Some(msg_str) = v { - - match serde_json::from_str::(msg_str.as_str()) { - Ok(waku_event) => { - println!("AAAAAA jamon {:?}", waku_event); - - // // match waku_event { - // // WakuEvent::WakuMessage(evt) => { - // // // println!("WakuMessage event received: {:?}", evt.waku_message); - - // // // match ::decode(evt.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(); - // // // } - // // // } - - // // }, - // // WakuEvent::Unrecognized(err) => println!("Unrecognized waku event: {:?}", err), - // // } - }, + let event: WakuEvent = + serde_json::from_str(v.unwrap().as_str()).expect("Parsing event to succeed"); - Err(_e) => {}, //eprintln!("Error reading file: {}", e), + match event { + WakuEvent::WakuMessage(evt) => { + + if evt.waku_message.content_topic != TOY_CHAT_CONTENT_TOPIC { + return; // skip the messages that don't belong to the toy chat + } + + match ::decode(evt.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(); + } + } } - } + WakuEvent::Unrecognized(err) => eprintln!("Unrecognized waku event: {:?}", err), + _ => eprintln!("event case not expected"), + }; } })?; @@ -161,77 +128,84 @@ impl App { } impl App { + + fn retrieve_history(&mut self) { + let history = self.waku.store_query(None, vec![TOY_CHAT_CONTENT_TOPIC.clone()], STORE_NODE); + let history = history.unwrap(); + + let messages = history.messages + .iter() + .map(|store_resp_msg| { + ::decode(store_resp_msg.message.payload()) + .expect("Toy chat messages should be decodeable") + }) + .collect(); + + if history.messages.len() > 0 { + *self.messages.write().unwrap() = messages; + } + } + fn run_main_loop( &mut self, terminal: &mut Terminal, ) -> std::result::Result<(), Box> { - // let history = self.waku.store_query(None, vec![TOY_CHAT_CONTENT_TOPIC.clone()], STORE_NODE); - // let history = history.unwrap(); + self.retrieve_history(); - // let messages = history.messages - // .iter() - // .map(|store_resp_msg| { - // ::decode(store_resp_msg.message.payload()) - // .expect("Toy chat messages should be decodeable") - // }) - // .collect(); - - // if history.messages.len() > 0 { - // *self.messages.write().unwrap() = messages; - // } - loop { terminal.draw(|f| ui(f, self))?; - if let Event::Key(key) = event::read()? { - match self.input_mode { - InputMode::Normal => match key.code { - KeyCode::Char('e') => { - self.input_mode = InputMode::Editing; - } - KeyCode::Char('q') => { - return Ok(()); - } - _ => {} - }, - InputMode::Editing => match key.code { - KeyCode::Enter => { - let message_content: String = self.input.drain(..).collect(); - let message = Chat2Message::new(&self.nick, &message_content); - let mut buff = Vec::new(); - let meta = Vec::new(); - Message::encode(&message, &mut buff)?; - let waku_message = WakuMessage::new( - buff, - TOY_CHAT_CONTENT_TOPIC.clone(), - 1, - Utc::now().timestamp_nanos() as usize, - meta, - false, - ); - - let pubsub_topic = PubsubTopic::new(DEFAULT_PUBSUB_TOPIC); - if let Err(e) = self.waku.relay_publish_message( - &waku_message, - &pubsub_topic, - None, - ) { - let mut out = std::io::stderr(); - write!(out, "{e:?}").unwrap(); + if event::poll(Duration::from_millis(500)).unwrap() { + if let Event::Key(key) = event::read()? { + match self.input_mode { + InputMode::Normal => match key.code { + KeyCode::Char('e') => { + self.input_mode = InputMode::Editing; } - } - KeyCode::Char(c) => { - self.input.push(c); - } - KeyCode::Backspace => { - self.input.pop(); - } - KeyCode::Esc => { - self.input_mode = InputMode::Normal; - } - _ => {} - }, + KeyCode::Char('q') => { + return Ok(()); + } + _ => {} + }, + InputMode::Editing => match key.code { + KeyCode::Enter => { + let message_content: String = self.input.drain(..).collect(); + let message = Chat2Message::new(&self.nick, &message_content); + let mut buff = Vec::new(); + let meta = Vec::new(); + Message::encode(&message, &mut buff)?; + let waku_message = WakuMessage::new( + buff, + TOY_CHAT_CONTENT_TOPIC.clone(), + 1, + Utc::now().timestamp_nanos() as usize, + meta, + false, + ); + + let pubsub_topic = PubsubTopic::new(DEFAULT_PUBSUB_TOPIC); + if let Err(e) = self.waku.relay_publish_message( + &waku_message, + &pubsub_topic, + None, + ) { + let mut out = std::io::stderr(); + write!(out, "{e:?}").unwrap(); + } + } + KeyCode::Char(c) => { + self.input.push(c); + } + KeyCode::Backspace => { + self.input.pop(); + } + KeyCode::Esc => { + self.input_mode = InputMode::Normal; + } + _ => {} + }, + } } } } @@ -242,47 +216,6 @@ impl App { } } -// fn retrieve_history( -// waku: &WakuNodeHandle, -// ) -> waku_bindings::Result> { -// let self_id = waku.peer_id().unwrap(); -// let peer = waku -// .peers()? -// .iter() -// .find(|&peer| peer.peer_id() != &self_id) -// .cloned() -// .unwrap(); - -// let result = waku.store_query( -// &StoreQuery { -// pubsub_topic: None, -// content_topics: vec![TOY_CHAT_CONTENT_TOPIC.clone()], -// start_time: Some( -// (Duration::from_secs(Utc::now().timestamp() as u64) -// - Duration::from_secs(60 * 60 * 24)) -// .as_nanos() as usize, -// ), -// end_time: None, -// paging_options: Some(PagingOptions { -// page_size: 25, -// cursor: None, -// forward: true, -// }), -// }, -// 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 main() -> std::result::Result<(), Box> { let nick = std::env::args().nth(1).expect("Nick to be set"); diff --git a/waku-bindings/src/general/mod.rs b/waku-bindings/src/general/mod.rs index 634de2d..4cb8524 100644 --- a/waku-bindings/src/general/mod.rs +++ b/waku-bindings/src/general/mod.rs @@ -26,22 +26,42 @@ pub struct WakuMessage { #[serde(with = "base64_serde", default = "Vec::new")] pub payload: Vec, /// The content topic to be set on the message - content_topic: WakuContentTopic, - // TODO: check if missing default should be 0 + pub content_topic: WakuContentTopic, /// The Waku Message version number #[serde(default)] - version: WakuMessageVersion, + pub version: WakuMessageVersion, /// Unix timestamp in nanoseconds #[serde(deserialize_with = "deserialize_number_from_string")] - timestamp: usize, - meta: Vec, + pub timestamp: usize, + #[serde(with = "base64_serde", default = "Vec::new")] + pub meta: Vec, #[serde(default)] - ephemeral: bool, + pub ephemeral: bool, // TODO: implement RLN fields #[serde(flatten)] _extras: serde_json::Value, } +#[derive(Clone, Serialize, Deserialize, Debug, Default)] +#[serde(rename_all = "camelCase")] +pub struct WakuStoreRespMessage { + // #[serde(with = "base64_serde", default = "Vec::new")] + pub payload: Vec, + /// The content topic to be set on the message + // #[serde(rename = "contentTopic")] + pub content_topic: String, + // #[serde(with = "base64_serde", default = "Vec::new")] + pub meta: Vec, + /// The Waku Message version number + #[serde(default)] + pub version: WakuMessageVersion, + /// Unix timestamp in nanoseconds + pub timestamp: usize, + #[serde(default)] + pub ephemeral: bool, + // pub proof: Vec, +} + impl WakuMessage { pub fn new, META: AsRef<[u8]>>( payload: PAYLOAD, @@ -70,6 +90,12 @@ impl WakuMessage { } } +impl WakuStoreRespMessage { + pub fn payload(&self) -> &[u8] { + &self.payload + } +} + mod base64_serde { use base64::Engine; use serde::de::Error; diff --git a/waku-bindings/src/node/config.rs b/waku-bindings/src/node/config.rs index 8d26f3b..68ca58e 100644 --- a/waku-bindings/src/node/config.rs +++ b/waku-bindings/src/node/config.rs @@ -33,6 +33,10 @@ pub struct WakuNodeConfig { #[serde(skip_serializing_if = "Option::is_none")] pub max_message_size: Option, + /// Store protocol + #[serde(skip_serializing_if = "Option::is_none")] + pub storenode: Option<&'static str>, + /// RLN configuration #[serde(skip_serializing_if = "Option::is_none")] pub rln_relay: Option, diff --git a/waku-bindings/src/node/mod.rs b/waku-bindings/src/node/mod.rs index 628ca2b..e52c28d 100644 --- a/waku-bindings/src/node/mod.rs +++ b/waku-bindings/src/node/mod.rs @@ -25,7 +25,6 @@ use crate::general::{MessageHash, Result, WakuMessage}; use crate::utils::LibwakuResponse; use crate::node::context::WakuNodeContext; -use crate::node::store::PagingOptions; pub use config::RLNConfig; pub use config::WakuNodeConfig; pub use events::{WakuEvent, WakuMessageEvent}; diff --git a/waku-bindings/src/node/store.rs b/waku-bindings/src/node/store.rs index 01b5d7e..a507c0a 100644 --- a/waku-bindings/src/node/store.rs +++ b/waku-bindings/src/node/store.rs @@ -7,10 +7,10 @@ use libc::*; // internal use crate::general::{ contenttopic::WakuContentTopic, pubsubtopic::PubsubTopic, MessageHash, Result, + WakuStoreRespMessage, }; use crate::node::context::WakuNodeContext; use crate::utils::{get_trampoline, handle_json_response, LibwakuResponse, WakuDecode}; -use crate::WakuMessage; use multiaddr::Multiaddr; use serde::{Deserialize, Serialize}; @@ -49,7 +49,7 @@ struct StoreQueryRequest { #[serde(rename_all = "camelCase")] pub struct StoreWakuMessageResponse { pub message_hash: [u8; 32], - pub message: WakuMessage, + pub message: WakuStoreRespMessage, pub pubsub_topic: String, } @@ -72,7 +72,7 @@ pub struct StoreResponse { // Implement WakuDecode for Vec impl WakuDecode for StoreResponse { fn decode(input: &str) -> Result { - let ret: StoreResponse = serde_json::from_str(input).expect("parse store resp correctly"); + let ret: StoreResponse = serde_json::from_str(input).expect("could not parse store resp"); Ok(ret) } }