Remove waku chat example.
This commit is contained in:
Alejandro Cabeza Romero
parent
2f708060f7
commit
b552e0e376
|
|
@ -1,19 +0,0 @@
|
|||
[package]
|
||||
name = "waku-chat"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
tokio = { version = "1", features = ["rt"] }
|
||||
waku-bindings = "0.3.1"
|
||||
serde = "1"
|
||||
bincode = "1"
|
||||
overwatch-rs = { path = "../../overwatch-rs" }
|
||||
overwatch-derive = { path = "../../overwatch-derive" }
|
||||
tracing = "*"
|
||||
async-trait = "0.1"
|
||||
tracing-subscriber = "0.3"
|
||||
clap = { version = "4.0.18", features = ["derive"] }
|
||||
rand = "0.8"
|
||||
|
|
@ -1,94 +0,0 @@
|
|||
use crate::network::*;
|
||||
use async_trait::async_trait;
|
||||
use overwatch_rs::services::handle::ServiceStateHandle;
|
||||
use overwatch_rs::services::relay::{NoMessage, OutboundRelay};
|
||||
use overwatch_rs::services::state::{NoOperator, NoState};
|
||||
use overwatch_rs::services::{ServiceCore, ServiceData, ServiceId};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::sync::mpsc::channel;
|
||||
|
||||
/// Chat service handler
|
||||
/// displays received messages, send new ones
|
||||
pub struct ChatService {
|
||||
service_state: ServiceStateHandle<Self>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize)]
|
||||
struct Message {
|
||||
user: usize,
|
||||
msg: Box<[u8]>,
|
||||
}
|
||||
|
||||
impl ServiceData for ChatService {
|
||||
const SERVICE_ID: ServiceId = "Chat";
|
||||
type Settings = usize;
|
||||
type State = NoState<Self::Settings>;
|
||||
type StateOperator = NoOperator<Self::State>;
|
||||
type Message = NoMessage;
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ServiceCore for ChatService {
|
||||
fn init(service_state: ServiceStateHandle<Self>) -> Result<Self, overwatch_rs::DynError> {
|
||||
Ok(Self { service_state })
|
||||
}
|
||||
|
||||
async fn run(self) -> Result<(), overwatch_rs::DynError> {
|
||||
let Self {
|
||||
mut service_state, ..
|
||||
} = self;
|
||||
// TODO: waku should not end up in the public interface of the network service, at least not as a type
|
||||
let mut network_relay = service_state
|
||||
.overwatch_handle
|
||||
.relay::<NetworkService<waku::Waku>>()
|
||||
.connect()
|
||||
.await
|
||||
.unwrap();
|
||||
let user = service_state.settings_reader.get_updated_settings();
|
||||
let (sender, mut receiver) = channel(1);
|
||||
// TODO: typestate so I can't call send if it's not connected
|
||||
network_relay
|
||||
.send(NetworkMsg::Subscribe {
|
||||
kind: EventKind::Message,
|
||||
sender,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// send new messages
|
||||
// for interactive stdin I/O it's recommended to
|
||||
// use an external thread, see https://docs.rs/tokio/latest/tokio/io/struct.Stdin.html
|
||||
std::thread::spawn(move || loop {
|
||||
let mut input = String::new();
|
||||
std::io::stdin()
|
||||
.read_line(&mut input)
|
||||
.expect("error reading message");
|
||||
input.truncate(input.trim().len());
|
||||
network_relay
|
||||
.blocking_send(NetworkMsg::Broadcast(
|
||||
bincode::serialize(&Message {
|
||||
user,
|
||||
msg: input.as_bytes().to_vec().into_boxed_slice(),
|
||||
})
|
||||
.unwrap()
|
||||
.into_boxed_slice(),
|
||||
))
|
||||
.unwrap();
|
||||
tracing::debug!("[sending]: {}...", input);
|
||||
});
|
||||
|
||||
// print received messages
|
||||
while let Some(NetworkEvent::RawMessage(message)) = receiver.recv().await {
|
||||
if let Ok(msg) = bincode::deserialize::<Message>(&message) {
|
||||
if msg.user != user {
|
||||
println!(
|
||||
"[received][{}]: {}",
|
||||
msg.user,
|
||||
String::from_utf8_lossy(&msg.msg)
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
|
@ -1,52 +0,0 @@
|
|||
#![allow(dead_code)]
|
||||
#![allow(unused)]
|
||||
// public chat service
|
||||
// messages are disseminated through waku,
|
||||
// no consensus, no blocks
|
||||
mod network;
|
||||
// TODO: different chat rooms with different contentTopicId
|
||||
mod chat;
|
||||
|
||||
use chat::*;
|
||||
use clap::Parser;
|
||||
use network::*;
|
||||
use overwatch_derive::*;
|
||||
use overwatch_rs::{overwatch::*, services::handle::ServiceHandle};
|
||||
|
||||
/// Simple program to greet a person
|
||||
#[derive(Parser, Debug)]
|
||||
#[command(author, version, about, long_about = None)]
|
||||
struct Args {
|
||||
/// Multiaddrs of other nodes participating in the protocol
|
||||
#[arg(short, long)]
|
||||
peers: Vec<String>,
|
||||
|
||||
/// Listening port
|
||||
port: u16,
|
||||
}
|
||||
|
||||
#[derive(Services)]
|
||||
struct Services {
|
||||
chat: ServiceHandle<ChatService>,
|
||||
network: ServiceHandle<NetworkService<network::waku::Waku>>,
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
fn main() {
|
||||
tracing_subscriber::fmt::init();
|
||||
let Args { peers, port } = Args::parse();
|
||||
let app = OverwatchRunner::<Services>::run(
|
||||
ServicesServiceSettings {
|
||||
chat: rand::random(),
|
||||
network: NetworkConfig { peers, port },
|
||||
},
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
app.wait_finished();
|
||||
}
|
||||
|
||||
#[cfg(not(target_os = "linux"))]
|
||||
fn main() {
|
||||
println!("waku is only supported on linux");
|
||||
}
|
||||
|
|
@ -1,82 +0,0 @@
|
|||
pub mod waku;
|
||||
use async_trait::async_trait;
|
||||
use overwatch_rs::services::handle::ServiceStateHandle;
|
||||
use overwatch_rs::services::relay::RelayMessage;
|
||||
use overwatch_rs::services::state::{NoOperator, NoState};
|
||||
use overwatch_rs::services::{ServiceCore, ServiceData, ServiceId};
|
||||
use std::fmt::Debug;
|
||||
use tokio::sync::mpsc::Sender;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum NetworkMsg {
|
||||
Broadcast(Box<[u8]>),
|
||||
Subscribe {
|
||||
kind: EventKind,
|
||||
sender: Sender<NetworkEvent>,
|
||||
},
|
||||
}
|
||||
|
||||
impl RelayMessage for NetworkMsg {}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum EventKind {
|
||||
Message,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum NetworkEvent {
|
||||
RawMessage(Box<[u8]>),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct NetworkConfig {
|
||||
pub port: u16,
|
||||
pub peers: Vec<String>,
|
||||
}
|
||||
|
||||
pub struct NetworkService<I: NetworkBackend + Send + 'static> {
|
||||
implem: I,
|
||||
service_state: ServiceStateHandle<Self>,
|
||||
}
|
||||
|
||||
impl<I: NetworkBackend + Send + 'static> ServiceData for NetworkService<I> {
|
||||
const SERVICE_ID: ServiceId = "Network";
|
||||
type Settings = NetworkConfig;
|
||||
type State = NoState<Self::Settings>;
|
||||
type StateOperator = NoOperator<Self::State>;
|
||||
type Message = NetworkMsg;
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl<I: NetworkBackend + Send + 'static> ServiceCore for NetworkService<I> {
|
||||
fn init(mut service_state: ServiceStateHandle<Self>) -> Result<Self, overwatch_rs::DynError> {
|
||||
Ok(Self {
|
||||
implem: <I as NetworkBackend>::new(
|
||||
service_state.settings_reader.get_updated_settings(),
|
||||
),
|
||||
service_state,
|
||||
})
|
||||
}
|
||||
|
||||
async fn run(mut self) -> Result<(), overwatch_rs::DynError> {
|
||||
let Self {
|
||||
service_state,
|
||||
mut implem,
|
||||
} = self;
|
||||
let mut relay = service_state.inbound_relay;
|
||||
|
||||
while let Some(msg) = relay.recv().await {
|
||||
match msg {
|
||||
NetworkMsg::Broadcast(msg) => implem.broadcast(msg),
|
||||
NetworkMsg::Subscribe { kind: _, sender } => implem.subscribe(sender),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub trait NetworkBackend {
|
||||
fn new(config: NetworkConfig) -> Self;
|
||||
fn broadcast(&self, msg: Box<[u8]>);
|
||||
fn subscribe(&mut self, sender: Sender<NetworkEvent>);
|
||||
}
|
||||
|
|
@ -1,78 +0,0 @@
|
|||
use super::*;
|
||||
use ::waku_bindings::*;
|
||||
use std::str::FromStr;
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::SystemTime;
|
||||
use tokio::sync::mpsc::Sender;
|
||||
|
||||
pub struct Waku {
|
||||
waku: WakuNodeHandle<Running>,
|
||||
subscribers: Arc<RwLock<Vec<Sender<NetworkEvent>>>>,
|
||||
}
|
||||
|
||||
impl NetworkBackend for Waku {
|
||||
fn new(config: NetworkConfig) -> Self {
|
||||
let waku_config = WakuNodeConfig {
|
||||
port: Some(config.port.into()),
|
||||
..Default::default()
|
||||
};
|
||||
let waku = waku_new(Some(waku_config)).unwrap().start().unwrap();
|
||||
for peer in config.peers {
|
||||
let addr = Multiaddr::from_str(&peer).unwrap();
|
||||
let peer_id = waku.add_peer(&addr, waku::ProtocolId::Relay).unwrap();
|
||||
waku.connect_peer_with_id(&peer_id, None).unwrap();
|
||||
}
|
||||
waku.relay_subscribe(None).unwrap();
|
||||
assert!(waku.relay_enough_peers(None).unwrap());
|
||||
tracing::info!("waku listening on {}", waku.listen_addresses().unwrap()[0]);
|
||||
Self {
|
||||
waku,
|
||||
subscribers: Arc::new(RwLock::new(Vec::new())),
|
||||
}
|
||||
}
|
||||
|
||||
fn subscribe(&mut self, sender: Sender<NetworkEvent>) {
|
||||
self.subscribers.write().unwrap().push(sender);
|
||||
tracing::debug!("someone subscribed");
|
||||
let subscribers = Arc::clone(&self.subscribers);
|
||||
waku_set_event_callback(move |sig| {
|
||||
match sig.event() {
|
||||
Event::WakuMessage(ref message_event) => {
|
||||
tracing::debug!("received message event");
|
||||
// we can probably avoid sending a copy to each subscriber and just borrow / clone on demand
|
||||
for s in subscribers.read().unwrap().iter() {
|
||||
s.try_send(NetworkEvent::RawMessage(
|
||||
message_event
|
||||
.waku_message()
|
||||
.payload()
|
||||
.to_vec()
|
||||
.into_boxed_slice(),
|
||||
))
|
||||
.unwrap()
|
||||
}
|
||||
}
|
||||
_ => tracing::debug!("unsupported event"),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn broadcast(&self, msg: Box<[u8]>) {
|
||||
let content_topic = WakuContentTopic::from_str("/waku/2/default-waku/proto").unwrap();
|
||||
let message = WakuMessage::new(
|
||||
msg,
|
||||
content_topic,
|
||||
1,
|
||||
SystemTime::now()
|
||||
.duration_since(SystemTime::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs() as usize,
|
||||
"",
|
||||
false,
|
||||
);
|
||||
let msg_id = self
|
||||
.waku
|
||||
.relay_publish_message(&message, None, None)
|
||||
.unwrap();
|
||||
tracing::debug!("sent msg {:?} with id {}", message.payload(), msg_id);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue