logos-co
/
Overwatch
mirror of https://github.com/logos-co/Overwatch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

chore(examples): Remove outdated waku example (#35)

This commit is contained in:
Álex 2024-12-19 16:37:17 +01:00 committed by GitHub
parent 2f708060f7 55def9fe94
commit 848e3635d7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 2 additions and 327 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -1 +1,3 @@
**target/
.idea/
Cargo.lock

1
Cargo.toml
View File

@ -3,7 +3,6 @@ resolver = "2"
members = [
"overwatch-rs",
"overwatch-derive",
"examples/waku-chat"
]
[profile.release-opt]

19
examples/waku-chat/Cargo.toml
View File

@ -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"

94
examples/waku-chat/src/chat.rs
View File

@ -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(())
}
}

52
examples/waku-chat/src/main.rs
View File

@ -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");
}

82
examples/waku-chat/src/network/mod.rs
View File

@ -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>);
}

78
examples/waku-chat/src/network/waku.rs
View File

@ -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);
}
}

1
overwatch-rs/src/services/handle.rs
View File

@ -127,7 +127,6 @@ where
{
/// Spawn the service main loop and handle it lifecycle
/// Return a handle to abort execution manually
pub fn run(self) -> Result<(ServiceId, LifecycleHandle), crate::DynError> {
let ServiceRunner {
service_state,