From 0c0699c0460c1b149915729223eec701bde481df Mon Sep 17 00:00:00 2001 From: hozan23 Date: Fri, 14 Jun 2024 02:04:43 +0200 Subject: p2p: WIP implement rpc server for the p2p monitor --- p2p/examples/monitor/src/main.rs | 240 +-------------------------------------- 1 file changed, 5 insertions(+), 235 deletions(-) (limited to 'p2p/examples/monitor/src/main.rs') diff --git a/p2p/examples/monitor/src/main.rs b/p2p/examples/monitor/src/main.rs index c57d06c..990f8d2 100644 --- a/p2p/examples/monitor/src/main.rs +++ b/p2p/examples/monitor/src/main.rs @@ -1,28 +1,21 @@ +mod service; mod shared; use std::sync::Arc; use clap::Parser; -use log::error; -use ringbuffer::{AllocRingBuffer, RingBuffer}; -use serde::{Deserialize, Serialize}; -use smol::{channel, lock::Mutex, Executor}; +use smol::{channel, Executor}; -use karyon_core::async_util::{CondWait, TaskGroup, TaskResult}; -use karyon_jsonrpc::{ - message::SubscriptionID, rpc_impl, rpc_pubsub_impl, Channel, Server, Subscription, -}; +use karyon_jsonrpc::Server; use karyon_p2p::{ endpoint::{Endpoint, Port}, keypair::{KeyPair, KeyPairType}, - monitor::{ConnEvent, DiscoveryEvent, PeerPoolEvent}, - ArcBackend, Backend, Config, Error, Result, + Backend, Config, }; +use service::MonitorRPC; use shared::run_executor; -const EVENT_BUFFER_SIZE: usize = 30; - #[derive(Parser)] #[command(author, version, about, long_about = None)] struct Cli { @@ -47,202 +40,6 @@ struct Cli { discovery_port: Option, } -struct MonitorRPC { - backend: ArcBackend, - conn_event_buffer: Arc>>, - pp_event_buffer: Arc>>, - discv_event_buffer: Arc>>, - conn_event_condvar: Arc, - pp_event_condvar: Arc, - discv_event_condvar: Arc, - task_group: TaskGroup, -} - -impl MonitorRPC { - fn new(backend: ArcBackend, ex: Arc>) -> Arc { - Arc::new(MonitorRPC { - backend, - conn_event_buffer: Arc::new(Mutex::new(AllocRingBuffer::new(EVENT_BUFFER_SIZE))), - pp_event_buffer: Arc::new(Mutex::new(AllocRingBuffer::new(EVENT_BUFFER_SIZE))), - discv_event_buffer: Arc::new(Mutex::new(AllocRingBuffer::new(EVENT_BUFFER_SIZE))), - conn_event_condvar: Arc::new(CondWait::new()), - pp_event_condvar: Arc::new(CondWait::new()), - discv_event_condvar: Arc::new(CondWait::new()), - task_group: TaskGroup::with_executor(ex.into()), - }) - } - - async fn run(&self) -> Result<()> { - let conn_events = self.backend.monitor().conn_events().await; - let peer_pool_events = self.backend.monitor().peer_pool_events().await; - let discovery_events = self.backend.monitor().discovery_events().await; - - let conn_event_buffer = self.conn_event_buffer.clone(); - let pp_event_buffer = self.pp_event_buffer.clone(); - let discv_event_buffer = self.discv_event_buffer.clone(); - - let conn_event_condvar = self.conn_event_condvar.clone(); - let pp_event_condvar = self.pp_event_condvar.clone(); - let discv_event_condvar = self.discv_event_condvar.clone(); - - let on_failuer = |res: TaskResult>| async move { - if let TaskResult::Completed(Err(err)) = res { - error!("Event receive loop: {err}") - } - }; - - self.task_group.spawn( - async move { - loop { - let event = conn_events.recv().await?; - conn_event_buffer.lock().await.push(event); - conn_event_condvar.broadcast().await; - } - }, - on_failuer, - ); - - self.task_group.spawn( - async move { - loop { - let event = peer_pool_events.recv().await?; - pp_event_buffer.lock().await.push(event); - pp_event_condvar.broadcast().await; - } - }, - on_failuer, - ); - - self.task_group.spawn( - async move { - loop { - let event = discovery_events.recv().await?; - discv_event_buffer.lock().await.push(event); - discv_event_condvar.broadcast().await; - } - }, - on_failuer, - ); - - Ok(()) - } - - async fn shutdown(&self) { - self.task_group.cancel().await; - } -} - -#[rpc_impl] -impl MonitorRPC { - async fn peer_id( - &self, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - Ok(serde_json::json!(self.backend.peer_id().to_string())) - } - - async fn inbound_connection( - &self, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - Ok(serde_json::json!(self.backend.inbound_slots())) - } - - async fn outbound_connection( - &self, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - Ok(serde_json::json!(self.backend.outbound_slots())) - } -} - -#[rpc_pubsub_impl] -impl MonitorRPC { - async fn conn_subscribe( - &self, - chan: Arc, - method: String, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub = chan.new_subscription(&method).await; - let sub_id = sub.id; - - let cond_wait = self.conn_event_condvar.clone(); - let buffer = self.conn_event_buffer.clone(); - self.task_group - .spawn(notify(sub, cond_wait, buffer), notify_failed); - - Ok(serde_json::json!(sub_id)) - } - - async fn peer_pool_subscribe( - &self, - chan: Arc, - method: String, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub = chan.new_subscription(&method).await; - let sub_id = sub.id; - - let cond_wait = self.pp_event_condvar.clone(); - let buffer = self.pp_event_buffer.clone(); - self.task_group - .spawn(notify(sub, cond_wait, buffer), notify_failed); - - Ok(serde_json::json!(sub_id)) - } - - async fn discovery_subscribe( - &self, - chan: Arc, - method: String, - _params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub = chan.new_subscription(&method).await; - let sub_id = sub.id; - - let cond_wait = self.discv_event_condvar.clone(); - let buffer = self.discv_event_buffer.clone(); - self.task_group - .spawn(notify(sub, cond_wait, buffer), notify_failed); - - Ok(serde_json::json!(sub_id)) - } - - async fn conn_unsubscribe( - &self, - chan: Arc, - _method: String, - params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub_id: SubscriptionID = serde_json::from_value(params)?; - chan.remove_subscription(&sub_id).await; - Ok(serde_json::json!(true)) - } - - async fn peer_pool_unsubscribe( - &self, - chan: Arc, - _method: String, - params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub_id: SubscriptionID = serde_json::from_value(params)?; - chan.remove_subscription(&sub_id).await; - Ok(serde_json::json!(true)) - } - - async fn discovery_unsubscribe( - &self, - chan: Arc, - _method: String, - params: serde_json::Value, - ) -> karyon_jsonrpc::Result { - let sub_id: SubscriptionID = serde_json::from_value(params)?; - chan.remove_subscription(&sub_id).await; - Ok(serde_json::json!(true)) - } -} - fn main() { env_logger::init(); @@ -309,30 +106,3 @@ fn main() { ex, ); } - -async fn notify + Clone>( - sub: Subscription, - cond_wait: Arc, - buffer: Arc>>, -) -> Result<()> { - for event in buffer.lock().await.iter() { - if let Err(err) = sub.notify(serde_json::json!(event)).await { - return Err(Error::Other(format!("failed to notify: {err}"))); - } - } - loop { - cond_wait.wait().await; - cond_wait.reset().await; - if let Some(event) = buffer.lock().await.back().cloned() { - if let Err(err) = sub.notify(serde_json::json!(event)).await { - return Err(Error::Other(format!("failed to notify: {err}"))); - } - } - } -} - -async fn notify_failed(result: TaskResult>) { - if let TaskResult::Completed(Err(err)) = result { - error!("Error: {err}"); - } -} -- cgit v1.2.3