waku-org
/
waku-rust-bindings
mirror of https://github.com/waku-org/waku-rust-bindings.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

some changes to add toy-chat example. not completed yet

This commit is contained in:
Ivan Folgueira Bande 2024-12-01 18:06:34 +01:00
parent ad0eab2c60
commit 0d37037d48
No known key found for this signature in database
GPG Key ID: 3C117481F89E24A7
17 changed files with 1513 additions and 445 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
Cargo.lock generated
View File

@ -530,8 +530,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bfd4d1b31faaa3a89d7934dbded3111da0d2ef28e3ebccdb4f0179f5929d1ef1"
dependencies = [
"iana-time-zone",
"js-sys",
"num-integer",
"num-traits",
"time",
"wasm-bindgen",
"winapi",
]
@ -1292,7 +1295,7 @@ checksum = "4eb1a864a501629691edf6c15a593b7a51eebaa1e8468e9ddc623de7c9b58ec6"
dependencies = [
"cfg-if",
"libc",
"wasi",
"wasi 0.11.0+wasi-snapshot-preview1",
]
[[package]]
@ -1695,7 +1698,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f3d0b296e374a4e6f3c7b0a1f5a51d748a0d34c85e7dc48fc3fa9a87657fe09"
dependencies = [
"libc",
"wasi",
"wasi 0.11.0+wasi-snapshot-preview1",
"windows-sys 0.48.0",
]
@ -2812,6 +2815,17 @@ dependencies = [
"once_cell",
]
[[package]]
name = "time"
version = "0.1.45"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b797afad3f312d1c66a56d11d0316f916356d11bd158fbc6ca6389ff6bf805a"
dependencies = [
"libc",
"wasi 0.10.0+wasi-snapshot-preview1",
"winapi",
]
[[package]]
name = "tiny-keccak"
version = "2.0.2"
@ -3053,6 +3067,7 @@ version = "0.5.0"
dependencies = [
"aes-gcm",
"base64 0.21.0",
"chrono",
"enr",
"futures",
"hex",
@ -3092,6 +3107,12 @@ dependencies = [
"winapi-util",
]
[[package]]
name = "wasi"
version = "0.10.0+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1a143597ca7c7793eff794def352d41792a93c481eb1042423ff7ff72ba2c31f"
[[package]]
name = "wasi"
version = "0.11.0+wasi-snapshot-preview1"

1221
examples/Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

3
examples/Cargo.toml
View File

@ -2,5 +2,6 @@
members = [
"basic",
"tic-tac-toe-gui"
"tic-tac-toe-gui",
"toy-chat"
]

16
examples/basic/src/main.rs
View File

@ -3,8 +3,8 @@ use std::str::from_utf8;
use std::time::SystemTime;
use tokio::time::{sleep, Duration};
use waku::{
general::pubsubtopic::PubsubTopic, waku_new, Encoding, Event, LibwakuResponse,
WakuContentTopic, WakuMessage, WakuNodeConfig,
general::pubsubtopic::PubsubTopic, waku_new, Encoding, LibwakuResponse, WakuContentTopic,
WakuEvent, WakuMessage, WakuNodeConfig,
};
#[tokio::main]
@ -26,11 +26,11 @@ async fn main() -> Result<(), Error> {
node2
.set_event_callback(|response| {
if let LibwakuResponse::Success(v) = response {
let event: Event =
let event: WakuEvent =
serde_json::from_str(v.unwrap().as_str()).expect("Parsing event to succeed");
match event {
Event::WakuMessage(evt) => {
WakuEvent::WakuMessage(evt) => {
println!("WakuMessage event received: {:?}", evt.waku_message);
let message = evt.waku_message;
let payload = message.payload.to_vec();
@ -39,7 +39,7 @@ async fn main() -> Result<(), Error> {
println!("Message Received in NODE 2: {}", msg);
println!("::::::::::::::::::::::::::::::::::::::::::::::::::::");
}
Event::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err),
WakuEvent::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err),
_ => panic!("event case not expected"),
};
}
@ -49,11 +49,11 @@ async fn main() -> Result<(), Error> {
node1
.set_event_callback(|response| {
if let LibwakuResponse::Success(v) = response {
let event: Event =
let event: WakuEvent =
serde_json::from_str(v.unwrap().as_str()).expect("Parsing event to succeed");
match event {
Event::WakuMessage(evt) => {
WakuEvent::WakuMessage(evt) => {
println!("WakuMessage event received: {:?}", evt.waku_message);
let message = evt.waku_message;
let payload = message.payload.to_vec();
@ -62,7 +62,7 @@ async fn main() -> Result<(), Error> {
println!("Message Received in NODE 1: {}", msg);
println!("::::::::::::::::::::::::::::::::::::::::::::::::::::");
}
Event::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err),
WakuEvent::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err),
_ => panic!("event case not expected"),
};
}

10
examples/tic-tac-toe-gui/src/main.rs
View File

@ -6,7 +6,7 @@ use std::time::{SystemTime, Duration};
use tokio::sync::mpsc;
use waku::{
waku_new, Encoding, Event, LibwakuResponse, WakuContentTopic,
waku_new, Encoding, WakuEvent, LibwakuResponse, WakuContentTopic,
WakuMessage, WakuNodeConfig, WakuNodeHandle, Initialized, Running,
general::pubsubtopic::PubsubTopic,
};
@ -53,11 +53,11 @@ impl TicTacToeApp<Initialized> {
let my_closure = move |response| {
if let LibwakuResponse::Success(v) = response {
let event: Event =
let event: WakuEvent =
serde_json::from_str(v.unwrap().as_str()).expect("Parsing event to succeed");
match event {
Event::WakuMessage(evt) => {
WakuEvent::WakuMessage(evt) => {
// println!("WakuMessage event received: {:?}", evt.waku_message);
let message = evt.waku_message;
let payload = message.payload.to_vec();
@ -74,7 +74,7 @@ impl TicTacToeApp<Initialized> {
}
}
}
Event::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err),
WakuEvent::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err),
_ => panic!("event case not expected"),
};
}
@ -83,8 +83,6 @@ impl TicTacToeApp<Initialized> {
// Establish a closure that handles the incoming messages
self.waku.set_event_callback(my_closure).expect("set event call back working");
let _ = self.waku.version();
// Start the waku node
let waku = self.waku.start().expect("waku should start");

18
examples/toy-chat/Cargo.toml Normal file
View File

@ -0,0 +1,18 @@
[package]
name = "toy-chat"
version = "0.1.0"
edition = "2021"
authors = [
"Daniel Sanchez Quiros <danielsq@status.im>"
]
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
waku = { path = "../../waku-bindings", package = "waku-bindings" }
serde_json = "1.0"
serde = { version = "1.0", features = ["derive"] }
tui = "0.19"
crossterm = "0.25"
unicode-width = "0.1"
prost = "0.11"
chrono = "0.4"

397
examples/toy-chat/src/main.rs Normal file
View File

@ -0,0 +1,397 @@
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::str::from_utf8;
use std::io::Write;
use std::sync::{Arc, RwLock};
use std::{error::Error, io};
use tui::{
backend::{Backend, CrosstermBackend},
layout::{Constraint, Direction, Layout},
style::{Color, Modifier, Style},
text::{Span, Spans, Text},
widgets::{Block, Borders, List, ListItem, Paragraph},
Frame, Terminal,
};
use unicode_width::UnicodeWidthStr;
use waku::{
general::pubsubtopic::PubsubTopic, general::Result, waku_new, Initialized, LibwakuResponse, Running, WakuEvent,
WakuMessage, WakuNodeConfig, WakuNodeHandle,
};
enum InputMode {
Normal,
Editing,
}
const STORE_NODE: &str = "/dns4/store-01.do-ams3.status.staging.status.im/tcp/30303/p2p/16Uiu2HAm3xVDaz6SRJ6kErwC21zBJEZjavVXg7VSkoWzaV1aMA3F";
const DEFAULT_PUBSUB_TOPIC: &str = "/waku/2/rs/16/32";
/// App holds the state of the application
struct App<State> {
/// Current value of the input box
input: String,
nick: String,
/// Current input mode
input_mode: InputMode,
/// History of recorded messages
messages: Arc<RwLock<Vec<Chat2Message>>>,
waku: WakuNodeHandle<State>,
}
impl App<Initialized> {
fn new(nick: String) -> Result<App<Initialized>> {
let pubsub_topic = PubsubTopic::new(DEFAULT_PUBSUB_TOPIC);
let waku = waku_new(Some(WakuNodeConfig {
tcp_port: Some(60010),
cluster_id: Some(16),
shards: vec![1, 32, 64, 128, 256],
// node_key: Some(SecretKey::from_str("2fc0515879e52b7b73297cfd6ab3abf7c344ef84b7a90ff6f4cc19e05a198027").unwrap()),
max_message_size: Some("1024KiB".to_string()),
relay_topics: vec![String::from(&pubsub_topic)],
log_level: Some("FATAL"), // Supported: TRACE, DEBUG, INFO, NOTICE, WARN, ERROR or FATAL
keep_alive: Some(true),
// Discovery
dns_discovery: Some(true),
dns_discovery_url: Some("enrtree://AMOJVZX4V6EXP7NTJPMAYJYST2QP6AJXYW76IU6VGJS7UVSNDYZG4@boot.prod.status.nodes.status.im"),
// discv5_discovery: Some(true),
// discv5_udp_port: Some(9001),
// discv5_enr_auto_update: Some(false),
..Default::default()
}))?;
Ok(App {
input: String::new(),
input_mode: InputMode::Normal,
messages: Arc::new(RwLock::new(Vec::new())),
nick: nick,
waku: waku,
})
}
fn start_waku_node(self) -> Result<App<Running>> {
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::<WakuEvent>(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 <Chat2Message as Message>::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),
// // }
},
Err(_e) => {}, //eprintln!("Error reading file: {}", e),
}
}
}
})?;
let waku = self.waku.start()?;
let pubsub_topic = PubsubTopic::new(DEFAULT_PUBSUB_TOPIC);
waku.relay_subscribe(&pubsub_topic)?;
Ok(App {
input: self.input,
nick: self.nick,
input_mode: self.input_mode,
messages: self.messages,
waku: waku,
})
}
}
impl App<Running> {
fn run_main_loop<B: Backend>(
&mut self,
terminal: &mut Terminal<B>,
) -> std::result::Result<(), Box<dyn Error>> {
// 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| {
// <Chat2Message as Message>::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();
}
}
KeyCode::Char(c) => {
self.input.push(c);
}
KeyCode::Backspace => {
self.input.pop();
}
KeyCode::Esc => {
self.input_mode = InputMode::Normal;
}
_ => {}
},
}
}
}
}
fn stop_app(self) {
self.waku.stop().expect("the node should stop properly");
}
}
// fn retrieve_history(
// waku: &WakuNodeHandle<Running>,
// ) -> waku_bindings::Result<Vec<Chat2Message>> {
// 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| {
// <Chat2Message as Message>::decode(waku_message.payload())
// .expect("Toy chat messages should be decodeable")
// })
// .collect())
// }
fn main() -> std::result::Result<(), Box<dyn Error>> {
let nick = std::env::args().nth(1).expect("Nick to be set");
let app = App::new(nick)?;
let mut app = app.start_waku_node()?;
// setup terminal
enable_raw_mode()?;
let mut stdout = io::stdout();
execute!(stdout, EnterAlternateScreen, EnableMouseCapture)?;
let backend = CrosstermBackend::new(stdout);
let mut terminal = Terminal::new(backend)?;
let res = app.run_main_loop(&mut terminal);
// restore terminal
disable_raw_mode()?;
execute!(
terminal.backend_mut(),
LeaveAlternateScreen,
DisableMouseCapture
)?;
terminal.show_cursor()?;
if let Err(err) = res {
println!("{err:?}")
}
Ok(())
}
fn ui<B: Backend, State>(f: &mut Frame<B>, app: &App<State>) {
let chunks = Layout::default()
.direction(Direction::Vertical)
.margin(2)
.constraints(
[
Constraint::Length(1),
Constraint::Length(3),
Constraint::Min(1),
]
.as_ref(),
)
.split(f.size());
let (msg, style) = match app.input_mode {
InputMode::Normal => (
vec![
Span::raw("Press "),
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 writing a message."),
],
Style::default().add_modifier(Modifier::RAPID_BLINK),
),
InputMode::Editing => (
vec![
Span::raw("Press "),
Span::styled("Esc", Style::default().add_modifier(Modifier::BOLD)),
Span::raw(" to stop editing, "),
Span::styled("Enter", Style::default().add_modifier(Modifier::BOLD)),
Span::raw(" to record the message"),
],
Style::default(),
),
};
let mut text = Text::from(Spans::from(msg));
text.patch_style(style);
let help_message = Paragraph::new(text);
f.render_widget(help_message, chunks[0]);
let input = Paragraph::new(app.input.as_ref())
.style(match app.input_mode {
InputMode::Normal => Style::default(),
InputMode::Editing => Style::default().fg(Color::Yellow),
})
.block(Block::default().borders(Borders::ALL).title("Input"));
f.render_widget(input, chunks[1]);
match app.input_mode {
InputMode::Normal =>
// Hide the cursor. `Frame` does this by default, so we don't need to do anything here
{}
InputMode::Editing => {
// Make the cursor visible and ask tui-rs to put it at the specified coordinates after rendering
f.set_cursor(
// Put cursor past the end of the input text
chunks[1].x + app.input.width() as u16 + 1,
// Move one line down, from the border to the input line
chunks[1].y + 1,
)
}
}
let messages: Vec<ListItem> = app
.messages
.read()
.unwrap()
.iter()
.map(|message| {
let content = vec![Spans::from(Span::raw(format!(
"[{} - {}]: {}",
message.timestamp().unwrap().format("%d-%m-%y %H:%M"),
message.nick(),
message.message()
)))];
ListItem::new(content)
})
.collect();
let messages = List::new(messages).block(Block::default().borders(Borders::ALL).title("Chat"));
f.render_widget(messages, chunks[2]);
}

37
examples/toy-chat/src/protocol.rs Normal file
View File

@ -0,0 +1,37 @@
use chrono::{DateTime, LocalResult, TimeZone, Utc};
use prost::Message;
use waku::{Encoding, WakuContentTopic};
pub static TOY_CHAT_CONTENT_TOPIC: WakuContentTopic =
WakuContentTopic::new("toy-chat", "2", "huilong", Encoding::Proto);
#[derive(Clone, Message)]
pub struct Chat2Message {
#[prost(uint64, tag = "1")]
timestamp: u64,
#[prost(string, tag = "2")]
nick: String,
#[prost(bytes, tag = "3")]
payload: Vec<u8>,
}
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) -> LocalResult<DateTime<Utc>> {
Utc.timestamp_opt(self.timestamp as i64, 0)
}
}

1
waku-bindings/Cargo.toml
View File

@ -33,6 +33,7 @@ serde-aux = "4.3.1"
rln = "0.3.4"
tokio = { version = "1", features = ["full"] }
regex = "1"
chrono = "0.4"
[dev-dependencies]
futures = "0.3.25"

10
waku-bindings/src/general/messagehash.rs Normal file
View File

@ -0,0 +1,10 @@
pub struct MessageHash {
pub data: [u8; 32],
}
impl MessageHash {
// Create a new hash with default (zeroed) data
pub fn new() -> Self {
MessageHash { data: [0u8; 32] }
}
}

18
waku-bindings/src/general/mod.rs
View File

@ -1,6 +1,7 @@
//! Waku [general](https://rfc.vac.dev/spec/36/#general) types
pub mod contenttopic;
pub mod messagehash;
pub mod pubsubtopic;
// crates
@ -20,21 +21,22 @@ pub type Result<T> = std::result::Result<T, String>;
/// as per the [specification](https://rfc.vac.dev/spec/36/#jsonmessage-type)
#[derive(Clone, Serialize, Deserialize, Debug, Default)]
#[serde(rename_all = "camelCase")]
pub struct WakuMessage {
#[serde(with = "base64_serde", default = "Vec::new")]
pub payload: Vec<u8>,
/// The content topic to be set on the message
pub content_topic: WakuContentTopic,
content_topic: WakuContentTopic,
// TODO: check if missing default should be 0
/// The Waku Message version number
#[serde(default)]
pub version: WakuMessageVersion,
version: WakuMessageVersion,
/// Unix timestamp in nanoseconds
#[serde(deserialize_with = "deserialize_number_from_string")]
pub timestamp: usize,
#[serde(with = "base64_serde", default = "Vec::new")]
pub meta: Vec<u8>,
timestamp: usize,
meta: Vec<u8>,
#[serde(default)]
pub ephemeral: bool,
ephemeral: bool,
// TODO: implement RLN fields
#[serde(flatten)]
_extras: serde_json::Value,
@ -62,6 +64,10 @@ impl WakuMessage {
_extras: Default::default(),
}
}
pub fn payload(&self) -> &[u8] {
&self.payload
}
}
mod base64_serde {

5
waku-bindings/src/general/pubsubtopic.rs
View File

@ -1,4 +1,7 @@
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PubsubTopic(String);
impl PubsubTopic {

2
waku-bindings/src/lib.rs
View File

@ -15,7 +15,7 @@ pub use utils::LibwakuResponse;
use rln;
pub use node::{
waku_create_content_topic, waku_new, Event, Initialized, Key, Multiaddr, PublicKey, RLNConfig,
waku_create_content_topic, waku_new, WakuEvent, Initialized, Key, Multiaddr, PublicKey, RLNConfig,
Running, SecretKey, WakuMessageEvent, WakuNodeConfig, WakuNodeHandle,
};

10
waku-bindings/src/node/events.rs
View File

@ -2,7 +2,7 @@
//!
//! Asynchronous events require a callback to be registered.
//! An example of an asynchronous event that might be emitted is receiving a message.
//! When an event is emitted, this callback will be triggered receiving an [`Event`]
//! When an event is emitted, this callback will be triggered receiving an [`WakuEvent`]
// crates
use serde::{Deserialize, Serialize};
@ -17,7 +17,7 @@ use crate::MessageHash;
#[non_exhaustive]
#[derive(Serialize, Deserialize, Debug)]
#[serde(tag = "eventType", rename_all = "camelCase")]
pub enum Event {
pub enum WakuEvent {
#[serde(rename = "message")]
WakuMessage(WakuMessageEvent),
Unrecognized(serde_json::Value),
@ -37,12 +37,12 @@ pub struct WakuMessageEvent {
#[cfg(test)]
mod tests {
use crate::Event;
use crate::WakuEvent;
#[test]
fn deserialize_message_event() {
let s = "{\"eventType\":\"message\",\"messageHash\":\"0x26ff3d7fbc950ea2158ce62fd76fd745eee0323c9eac23d0713843b0f04ea27c\",\"pubsubTopic\":\"/waku/2/default-waku/proto\",\"wakuMessage\":{\"payload\":\"SGkgZnJvbSDwn6aAIQ==\",\"contentTopic\":\"/toychat/2/huilong/proto\",\"timestamp\":1665580926660}}";
let evt: Event = serde_json::from_str(s).unwrap();
assert!(matches!(evt, Event::WakuMessage(_)));
let evt: WakuEvent = serde_json::from_str(s).unwrap();
assert!(matches!(evt, WakuEvent::WakuMessage(_)));
}
}

33
waku-bindings/src/node/mod.rs
View File

@ -8,13 +8,16 @@ mod lightpush;
mod management;
mod peers;
mod relay;
mod store;
// std
pub use aes_gcm::Key;
use chrono::Utc;
pub use multiaddr::Multiaddr;
pub use secp256k1::{PublicKey, SecretKey};
use std::marker::PhantomData;
use std::time::Duration;
use store::StoreResponse;
// internal
use crate::general::contenttopic::{Encoding, WakuContentTopic};
pub use crate::general::pubsubtopic::PubsubTopic;
@ -22,9 +25,10 @@ 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::{Event, WakuMessageEvent};
pub use events::{WakuEvent, WakuMessageEvent};
pub use relay::waku_create_content_topic;
use std::time::SystemTime;
@ -178,4 +182,31 @@ impl WakuNodeHandle<Running> {
) -> Result<MessageHash> {
lightpush::waku_lightpush_publish_message(&self.ctx, message, pubsub_topic)
}
pub fn store_query(
&self,
pubsub_topic: Option<PubsubTopic>,
content_topics: Vec<WakuContentTopic>,
peer_addr: &str,
) -> Result<StoreResponse> {
store::waku_store_query(
&self.ctx,
"hard-coded-req-id".to_string(),
true, // include_data
pubsub_topic,
content_topics,
Some(
(Duration::from_secs(Utc::now().timestamp() as u64)
- Duration::from_secs(60 * 60 * 24))
.as_nanos() as usize,
), // time_start
None, // end_time
None, // message_hashes
None, // pagination_cursor
true, // pagination_forward
Some(25), // pagination_limit,
peer_addr,
None, // timeout_millis
)
}
}

144
waku-bindings/src/node/store.rs Normal file
View File

@ -0,0 +1,144 @@
//! Waku store protocol related methods
// std
use std::ffi::CString;
// crates
use libc::*;
// internal
use crate::general::{
contenttopic::WakuContentTopic, pubsubtopic::PubsubTopic, MessageHash, Result,
};
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};
#[derive(Clone, Serialize, Deserialize, Debug)]
#[serde(rename_all = "camelCase")]
pub struct PagingOptions {
pub page_size: usize,
pub cursor: Option<MessageHash>,
pub forward: bool,
}
/// Criteria used to retrieve historical messages
#[derive(Clone, Serialize, Debug)]
struct StoreQueryRequest {
/// if true, the store-response will include the full message content. If false,
/// the store-response will only include a list of message hashes.
request_id: String,
include_data: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pubsub_topic: Option<PubsubTopic>,
content_topics: Vec<WakuContentTopic>,
#[serde(skip_serializing_if = "Option::is_none")]
time_start: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
time_end: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
message_hashes: Option<Vec<MessageHash>>,
#[serde(skip_serializing_if = "Option::is_none")]
pagination_cursor: Option<MessageHash>, // Message hash (key) from where to start query (exclusive)
pagination_forward: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pagination_limit: Option<u64>,
}
#[derive(Clone, Deserialize, Debug)]
#[serde(rename_all = "camelCase")]
pub struct StoreWakuMessageResponse {
pub message_hash: [u8; 32],
pub message: WakuMessage,
pub pubsub_topic: String,
}
#[derive(Clone, Deserialize, Debug)]
#[serde(rename_all = "camelCase")]
pub struct StoreResponse {
pub request_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub status_code: u32,
pub status_desc: String,
/// Array of retrieved historical messages in [`WakuMessage`] format
// #[serde(default)]
pub messages: Vec<StoreWakuMessageResponse>,
/// Paging information in [`PagingOptions`] format from which to resume further historical queries
#[serde(skip_serializing_if = "Option::is_none")]
pub pagination_cursor: Option<[u8; 32]>,
}
// Implement WakuDecode for Vec<Multiaddr>
impl WakuDecode for StoreResponse {
fn decode(input: &str) -> Result<Self> {
let ret: StoreResponse = serde_json::from_str(input).expect("parse store resp correctly");
Ok(ret)
}
}
pub fn waku_store_query(
ctx: &WakuNodeContext,
request_id: String,
include_data: bool,
pubsub_topic: Option<PubsubTopic>,
content_topics: Vec<WakuContentTopic>,
time_start: Option<usize>,
time_end: Option<usize>,
message_hashes: Option<Vec<MessageHash>>,
pagination_cursor: Option<MessageHash>, // Message hash (key) from where to start query (exclusive)
pagination_forward: bool,
pagination_limit: Option<u64>,
peer_addr: &str,
timeout_millis: Option<i32>,
) -> Result<StoreResponse> {
let query = StoreQueryRequest {
request_id,
include_data,
pubsub_topic,
content_topics,
time_start,
time_end,
message_hashes,
pagination_cursor,
pagination_forward,
pagination_limit,
};
let json_query = CString::new(
serde_json::to_string(&query).expect("StoreQuery should always be able to be serialized"),
)
.expect("CString should build properly from the serialized filter subscription")
.into_raw();
peer_addr
.parse::<Multiaddr>()
.expect("correct multiaddress in store query");
let peer_addr = CString::new(peer_addr)
.expect("peer_addr CString should be created")
.into_raw();
let timeout_millis = timeout_millis.unwrap_or(10000i32);
let mut result: LibwakuResponse = Default::default();
let result_cb = |r: LibwakuResponse| result = r;
let code = unsafe {
let mut closure = result_cb;
let cb = get_trampoline(&closure);
let out = waku_sys::waku_store_query(
ctx.get_ptr(),
json_query,
peer_addr,
timeout_millis,
cb,
&mut closure as *mut _ as *mut c_void,
);
drop(CString::from_raw(json_query));
drop(CString::from_raw(peer_addr));
out
};
handle_json_response(code, result)
}

8
waku-bindings/tests/node.rs
View File

@ -10,7 +10,7 @@ use tokio::time;
use tokio::time::sleep;
use waku_bindings::node::PubsubTopic;
use waku_bindings::{
waku_new, Encoding, Event, Initialized, MessageHash, WakuContentTopic, WakuMessage,
waku_new, Encoding, WakuEvent, Initialized, MessageHash, WakuContentTopic, WakuMessage,
WakuNodeConfig, WakuNodeHandle,
};
use waku_bindings::{LibwakuResponse, Running};
@ -45,16 +45,16 @@ async fn test_echo_messages(
let rx_waku_message_cloned = rx_waku_message.clone();
let closure = move |response| {
if let LibwakuResponse::Success(v) = response {
let event: Event =
let event: WakuEvent =
serde_json::from_str(v.unwrap().as_str()).expect("Parsing event to succeed");
match event {
Event::WakuMessage(evt) => {
WakuEvent::WakuMessage(evt) => {
if let Ok(mut msg_lock) = rx_waku_message_cloned.lock() {
*msg_lock = evt.waku_message;
}
}
Event::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err),
WakuEvent::Unrecognized(err) => panic!("Unrecognized waku event: {:?}", err),
_ => panic!("event case not expected"),
};
}