p2p: debug net info cleanup and add support to darkirc

bin/darkirc/src/main.rs

@@ -40,7 +40,7 @@ use darkfi::{
         view::View,
     },
     net,
-    rpc::server::listen_and_serve,
+    rpc::{jsonrpc::JsonSubscriber, server::listen_and_serve},
     system::{Subscriber, SubscriberPtr},
     util::{async_util::sleep, file::save_json_file, path::expand_path, time::Timestamp},
     Result,
@@ -208,24 +208,47 @@ async fn realmain(settings: Args, executor: Arc<smol::Executor<'_>>) -> Result<(
         })
         .await;
 
-    // Start
-    p2p.clone().start(executor.clone()).await?;
-
-    // Run
-    let executor_cloned = executor.clone();
-    executor_cloned.spawn(p2p.clone().run(executor.clone())).detach();
+    // Here we initialize the subscribers for dnetview streamed notifications
+    let json_subscriber = JsonSubscriber::new("dnet.subscribe_events");
+    let json_subscriber2 = json_subscriber.clone();
+    // Grab events from dnet subsystem
+    let dnet_subscriber = p2p.dnet_subscribe().await;
+    // Now join events coming from dnet_subscriber into json_subscriber
+    executor
+        .spawn(async move {
+            println!("here!!!!");
+            loop {
+                let event = dnet_subscriber.receive().await;
+                println!("event received!");
+                // Convert event to JSON
+                json_subscriber2.notify(&[event]).await;
+            }
+        })
+        .detach();
 
     ////////////////////
     // RPC interface setup
     ////////////////////
     let rpc_listen_addr = settings.rpc_listen.clone();
-    let rpc_interface =
-        Arc::new(JsonRpcInterface { addr: rpc_listen_addr.clone(), p2p: p2p.clone() });
+    let rpc_interface = Arc::new(JsonRpcInterface {
+        addr: rpc_listen_addr.clone(),
+        p2p: p2p.clone(),
+        dnet_subscriber: json_subscriber,
+    });
     let _ex = executor.clone();
     executor
         .spawn(async move { listen_and_serve(rpc_listen_addr, rpc_interface, _ex).await })
         .detach();
 
+    ////////////////////
+    // Start P2P network
+    ////////////////////
+    p2p.clone().start(executor.clone()).await?;
+
+    // Run
+    let executor_cloned = executor.clone();
+    executor_cloned.spawn(p2p.clone().run(executor.clone())).detach();
+
     ////////////////////
     // IRC server
     ////////////////////

bin/darkirc/src/rpc.rs

@@ -24,7 +24,7 @@ use url::Url;
 use darkfi::{
     net,
     rpc::{
-        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult},
+        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResponse, JsonResult, JsonSubscriber},
         server::RequestHandler,
     },
 };
@@ -32,6 +32,7 @@ use darkfi::{
 pub struct JsonRpcInterface {
     pub addr: Url,
     pub p2p: net::P2pPtr,
+    pub dnet_subscriber: JsonSubscriber,
 }
 
 #[async_trait]
@@ -41,7 +42,8 @@ impl RequestHandler for JsonRpcInterface {
 
         match req.method.as_str() {
             "ping" => self.pong(req.id, req.params).await,
-            "dnet_switch" => self.dnet_switch(req.id, req.params).await,
+            "dnet.switch" => self.dnet_switch(req.id, req.params).await,
+            "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
             _ => JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         }
     }
@@ -53,11 +55,11 @@ impl JsonRpcInterface {
     // By sending `true`, dnet will be activated, and by sending `false` dnet
     // will be deactivated. Returns `true` on success.
     //
-    // --> {"jsonrpc": "2.0", "method": "dnet_switch", "params": [true], "id": 42}
+    // --> {"jsonrpc": "2.0", "method": "dnet.switch", "params": [true], "id": 42}
     // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
     async fn dnet_switch(&self, id: u16, params: JsonValue) -> JsonResult {
         let params = params.get::<Vec<JsonValue>>().unwrap();
-        if params.len() != 1 || params[0].is_bool() {
+        if params.len() != 1 || !params[0].is_bool() {
             return JsonError::new(ErrorCode::InvalidParams, None, id).into()
         }
 
@@ -71,4 +73,20 @@ impl JsonRpcInterface {
 
         JsonResponse::new(JsonValue::Boolean(true), id).into()
     }
+
+    // RPCAPI:
+    // Initializes a subscription to p2p dnet events.
+    // Once a subscription is established, `darkirc` will send JSON-RPC notifications of
+    // new network events to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "dnet.subscribe_events", "params": [`event`]}
+    pub async fn dnet_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if !params.is_empty() {
+            return JsonError::new(ErrorCode::InvalidParams, None, id).into()
+        }
+
+        self.dnet_subscriber.clone().into()
+    }
 }

script/nodetool.py

@@ -12,7 +12,8 @@ class JsonRpc:
         await self.writer.wait_closed()
 
     async def _make_request(self, method, params):
-        ident = random.randint(0, 2**32)
+        ident = random.randint(0, 2**16)
+        print(ident)
         request = {
             "jsonrpc": "2.0",
             "method": method,
@@ -27,23 +28,31 @@ class JsonRpc:
         data = await self.reader.readline()
         message = data.decode()
         response = json.loads(message)
-        return response["result"]
+        print(response)
+        return "hello"
+        #return response["result"]
 
     async def ping(self):
         return await self._make_request("ping", [])
 
     async def dnet_switch(self, state):
-        return await self._make_request("dnet_switch", [state])
+        return await self._make_request("dnet.switch", [state])
 
-    async def dnet_info(self):
-        return await self._make_request("dnet_info", [])
+    async def dnet_subscribe_events(self):
+        return await self._make_request("dnet.subscribe_events", [])
+
+    #async def dnet_info(self):
+    #    return await self._make_request("dnet_info", [])
 
 async def main(argv):
     rpc = JsonRpc()
     await rpc.start("localhost", 26660)
     await rpc.dnet_switch(True)
+    await rpc.dnet_subscribe_events()
 
-    print(await rpc.dnet_info())
+    while True:
+        data = await rpc.reader.readline()
+        #print(await rpc.dnet_info())
 
     await rpc.dnet_switch(False)
     await rpc.stop()

src/net/channel.rs

@@ -17,7 +17,7 @@
  */
 
 use async_std::sync::{Arc, Mutex};
-use darkfi_serial::serialize;
+use darkfi_serial::{serialize, SerialDecodable, SerialEncodable};
 use futures::{
     io::{ReadHalf, WriteHalf},
     AsyncReadExt,
@@ -28,10 +28,11 @@ use smol::Executor;
 use url::Url;
 
 use super::{
+    dnet::{dnet, DnetEvent, MessageInfo},
     message,
     message::Packet,
     message_subscriber::{MessageSubscription, MessageSubsystem},
-    p2p::{dnet, P2pPtr},
+    p2p::P2pPtr,
     session::{Session, SessionBitFlag, SessionWeakPtr},
     transport::PtStream,
 };
@@ -45,26 +46,15 @@ use crate::{
 pub type ChannelPtr = Arc<Channel>;
 
 /// Channel debug info
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
 pub struct ChannelInfo {
-    pub addr: Url,
+    pub address: Url,
     pub random_id: usize,
-    pub remote_node_id: String,
-    pub time: NanoTimestamp,
-    pub op: String,
-    pub cmd: String,
 }
 
 impl ChannelInfo {
-    fn new(addr: Url) -> Self {
-        Self {
-            addr,
-            random_id: OsRng.gen(),
-            remote_node_id: String::new(),
-            time: NanoTimestamp::current_time(),
-            op: String::new(),
-            cmd: String::new(),
-        }
+    fn new(address: Url) -> Self {
+        Self { address, random_id: OsRng.gen() }
     }
 }
 
@@ -74,8 +64,6 @@ pub struct Channel {
     reader: Mutex<ReadHalf<Box<dyn PtStream>>>,
     /// The writing half of the transport stream
     writer: Mutex<WriteHalf<Box<dyn PtStream>>>,
-    /// Socket address
-    address: Url,
     /// The message subsystem instance for this channel
     message_subsystem: MessageSubsystem,
     /// Subscriber listening for stop signal for closing this channel
@@ -87,12 +75,12 @@ pub struct Channel {
     /// Weak pointer to respective session
     session: SessionWeakPtr,
     /// Channel debug info
-    info: Mutex<ChannelInfo>,
+    info: ChannelInfo,
 }
 
 impl std::fmt::Debug for Channel {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.address)
+        write!(f, "{}", self.address())
     }
 }
 
@@ -112,12 +100,11 @@ impl Channel {
         let message_subsystem = MessageSubsystem::new();
         Self::setup_dispatchers(&message_subsystem).await;
 
-        let info = Mutex::new(ChannelInfo::new(address.clone()));
+        let info = ChannelInfo::new(address.clone());
 
         Arc::new(Self {
             reader,
             writer,
-            address,
             message_subsystem,
             stop_subscriber: Subscriber::new(),
             receive_task: StoppableTask::new(),
@@ -225,11 +212,12 @@ impl Channel {
         let packet = Packet { command: M::NAME.to_string(), payload: serialize(message) };
 
         dnet!(self,
-            let mut info = self.info.lock().await;
-            info.time = NanoTimestamp::current_time();
-            info.op = "send".to_string();
-            info.cmd = packet.command.clone();
-            self.p2p().dnet_sub().notify(info.clone()).await;
+            let event = DnetEvent::SendMessage(MessageInfo {
+                chan: self.info.clone(),
+                cmd: packet.command.clone(),
+                time: NanoTimestamp::current_time(),
+            });
+            self.p2p().dnet_notify(event).await;
         );
 
         let stream = &mut *self.writer.lock().await;
@@ -311,7 +299,7 @@ impl Channel {
 
     /// Returns the local socket address
     pub fn address(&self) -> &Url {
-        &self.address
+        &self.info.address
     }
 
     /// Returns the inner [`MessageSubsystem`] reference

src/net/dnet.rs

@@ -0,0 +1,26 @@
+use super::channel::ChannelInfo;
+use crate::util::time::NanoTimestamp;
+use darkfi_serial::{SerialDecodable, SerialEncodable};
+
+macro_rules! dnet {
+    ($self:expr, $($code:tt)*) => {
+        {
+            if *$self.p2p().dnet_enabled.lock().await {
+                $($code)*
+            }
+        }
+    };
+}
+pub(crate) use dnet;
+
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct MessageInfo {
+    pub chan: ChannelInfo,
+    pub cmd: String,
+    pub time: NanoTimestamp,
+}
+
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub enum DnetEvent {
+    SendMessage(MessageInfo),
+}

src/net/mod.rs

@@ -70,7 +70,6 @@ pub use channel::ChannelPtr;
 /// The channel store is a hashmap of channel addresses that we can use
 /// to add and remove channels or check whether a channel is already in
 /// the store.
-#[macro_use]
 pub mod p2p;
 pub use p2p::{P2p, P2pPtr};
 
@@ -117,3 +116,8 @@ pub mod connector;
 /// behaviour and is controlled by clients of this API.
 pub mod settings;
 pub use settings::Settings;
+
+/// Optional events based debug-notify subsystem. Off by default. Enabled in P2P instance,
+/// and then call `p2p.dnet_sub()` to start receiving events.
+#[macro_use]
+pub mod dnet;

src/net/p2p.rs

@@ -29,7 +29,8 @@ use smol::Executor;
 use url::Url;
 
 use super::{
-    channel::{ChannelInfo, ChannelPtr},
+    channel::ChannelPtr,
+    dnet::DnetEvent,
     hosts::{Hosts, HostsPtr},
     message::Message,
     protocol::{protocol_registry::ProtocolRegistry, register_default_protocols},
@@ -80,7 +81,7 @@ pub struct P2p {
     /// Enable network debugging
     pub dnet_enabled: Mutex<bool>,
     /// The subscriber for which we can give dnet info over
-    dnet_sub: SubscriberPtr<ChannelInfo>,
+    dnet_sub: SubscriberPtr<DnetEvent>,
 }
 
 impl P2p {
@@ -319,18 +320,10 @@ impl P2p {
     }
 
     /// Return a reference to the dnet subscriber
-    pub fn dnet_sub(&self) -> SubscriberPtr<ChannelInfo> {
-        self.dnet_sub.clone()
+    pub async fn dnet_subscribe(&self) -> Subscription<DnetEvent> {
+        self.dnet_sub.clone().subscribe().await
+    }
+    pub async fn dnet_notify(&self, event: DnetEvent) {
+        self.dnet_sub.notify(event).await;
     }
 }
-
-macro_rules! dnet {
-    ($self:expr, $($code:tt)*) => {
-        {
-            if *$self.p2p().dnet_enabled.lock().await {
-                $($code)*
-            }
-        }
-    };
-}
-pub(crate) use dnet;