add get_info() hooks throughout net code (incomplete) used for introspecting state while p2p network is live.

bin/dnetview/src/main.rs

@@ -267,7 +267,7 @@ async fn render<B: Backend>(terminal: &mut Terminal<B>, model: Arc<Model>) -> io
             match k.unwrap() {
                 Key::Char('q') => {
                     terminal.clear()?;
-                    return Ok(());
+                    return Ok(())
                 }
                 Key::Char('j') => {
                     view.id_list.next();

src/net/channel.rs

@@ -3,6 +3,7 @@ use futures::{
     io::{ReadHalf, WriteHalf},
     AsyncReadExt,
 };
+use serde_json::json;
 use std::{
     net::{SocketAddr, TcpStream},
     sync::{
@@ -11,7 +12,7 @@ use std::{
     },
 };
 
-use log::*;
+use log::{debug, error, info};
 use smol::{Async, Executor};
 
 use crate::{
@@ -26,6 +27,24 @@ use crate::{
 /// Atomic pointer to async channel.
 pub type ChannelPtr = Arc<Channel>;
 
+struct ChannelInfo {
+    last_msg: String,
+    last_status: String,
+}
+
+impl ChannelInfo {
+    fn new() -> Self {
+        Self { last_msg: String::new(), last_status: String::new() }
+    }
+
+    async fn get_info(&self) -> serde_json::Value {
+        json!({
+            "last_msg": self.last_msg,
+            "last_status": self.last_status,
+        })
+    }
+}
+
 /// Async channel for communication between nodes.
 pub struct Channel {
     reader: Mutex<ReadHalf<Async<TcpStream>>>,
@@ -35,6 +54,7 @@ pub struct Channel {
     stop_subscriber: SubscriberPtr<Error>,
     receive_task: StoppableTaskPtr,
     stopped: AtomicBool,
+    info: Mutex<ChannelInfo>,
 }
 
 impl Channel {
@@ -57,9 +77,14 @@ impl Channel {
             stop_subscriber: Subscriber::new(),
             receive_task: StoppableTask::new(),
             stopped: AtomicBool::new(false),
+            info: Mutex::new(ChannelInfo::new()),
         })
     }
 
+    pub async fn get_info(&self) -> serde_json::Value {
+        self.info.lock().await.get_info().await
+    }
+
     /// Starts the channel. Runs a receive loop to start receiving messages or
     /// handles a network failure.
     pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
@@ -127,11 +152,18 @@ impl Channel {
                 Err(Error::ChannelStopped)
             }
         };
+
         debug!(target: "net",
             "Channel::send() [END, command={:?}, address={}]",
             M::name(),
             self.address()
         );
+        {
+            let info = &mut *self.info.lock().await;
+            info.last_msg = M::name().to_string();
+            info.last_status = "sent".to_string();
+        }
+
         result
     }
 
@@ -219,6 +251,11 @@ impl Channel {
                     return Err(Error::ChannelStopped)
                 }
             };
+            {
+                let info = &mut *self.info.lock().await;
+                info.last_msg = packet.command.clone();
+                info.last_status = "recv".to_string();
+            }
 
             // Send result to our subscribers
             self.message_subsystem.notify(&packet.command, packet.payload).await;

src/net/p2p.rs

@@ -26,6 +26,28 @@ pub type ConnectedChannels<T> = Mutex<HashMap<SocketAddr, Arc<T>>>;
 /// Atomic pointer to p2p interface.
 pub type P2pPtr = Arc<P2p>;
 
+enum P2pState {
+    // The p2p object has been created but not yet started.
+    Open,
+    // We are performing the initial seed session
+    Start,
+    // Seed session finished, but not yet running
+    Started,
+    // p2p is running and the network is active.
+    Run,
+}
+
+impl P2pState {
+    fn to_string(&self) -> String {
+        match self {
+            Self::Open => "open".to_string(),
+            Self::Start => "start".to_string(),
+            Self::Started => "started".to_string(),
+            Self::Run => "run".to_string(),
+        }
+    }
+}
+
 /// Top level peer-to-peer networking interface.
 pub struct P2p {
     pending: PendingChannels,
@@ -41,6 +63,8 @@ pub struct P2p {
     session_inbound: Mutex<Option<Arc<InboundSession>>>,
     session_outbound: Mutex<Option<Arc<OutboundSession>>>,
 
+    state: Mutex<P2pState>,
+
     settings: SettingsPtr,
 }
 
@@ -49,7 +73,7 @@ impl P2p {
     pub async fn new(settings: Settings) -> Arc<Self> {
         let settings = Arc::new(settings);
 
-        let mut self_ = Arc::new(Self {
+        let self_ = Arc::new(Self {
             pending: Mutex::new(HashSet::new()),
             channels: Mutex::new(HashMap::new()),
             channel_subscriber: Subscriber::new(),
@@ -59,6 +83,7 @@ impl P2p {
             session_manual: Mutex::new(None),
             session_inbound: Mutex::new(None),
             session_outbound: Mutex::new(None),
+            state: Mutex::new(P2pState::Open),
             settings,
         });
 
@@ -74,16 +99,18 @@ impl P2p {
     }
 
     pub async fn get_info(&self) -> serde_json::Value {
+        let external_addr = self
+            .settings
+            .external_addr
+            .map(|addr| serde_json::Value::from(addr.to_string()))
+            .unwrap_or(serde_json::Value::Null);
+
         json!({
-            "session_manual": self.session_manual().await.get_info(),
-            "session_inbound": self.session_inbound().await.get_info(),
-            "session_outbound": self.session_inbound().await.get_info(),
-            // Possible states:
-            //   open - the p2p object has been created but not yet started.
-            //   start - we are performing the initial seed session
-            //   started - seed session finished, but not yet running
-            //   run - p2p is running and the network is active.
-            "state": "open",
+            "external_addr": external_addr,
+            "session_manual": self.session_manual().await.get_info().await,
+            "session_inbound": self.session_inbound().await.get_info().await,
+            "session_outbound": self.session_inbound().await.get_info().await,
+            "state": self.state.lock().await.to_string(),
         })
     }
 
@@ -91,11 +118,15 @@ impl P2p {
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net", "P2p::start() [BEGIN]");
 
+        *self.state.lock().await = P2pState::Start;
+
         // Start seed session
         let seed = SeedSession::new(Arc::downgrade(&self));
         // This will block until all seed queries have finished
         seed.start(executor.clone()).await?;
 
+        *self.state.lock().await = P2pState::Started;
+
         debug!(target: "net", "P2p::start() [END]");
         Ok(())
     }
@@ -115,6 +146,8 @@ impl P2p {
     pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net", "P2p::run() [BEGIN]");
 
+        *self.state.lock().await = P2pState::Run;
+
         let manual = self.session_manual().await;
         for peer in &self.settings.peers {
             manual.clone().connect(peer, executor.clone()).await;

src/net/session/inbound_session.rs

@@ -1,3 +1,4 @@
+use async_trait::async_trait;
 use serde_json::{json, Value};
 use std::{
     net::SocketAddr,
@@ -120,8 +121,9 @@ impl InboundSession {
     }*/
 }
 
+#[async_trait]
 impl Session for InboundSession {
-    fn get_info(&self) -> serde_json::Value {
+    async fn get_info(&self) -> serde_json::Value {
         json!({
             "key": 110
         })

src/net/session/manual_session.rs

@@ -1,4 +1,5 @@
 use async_std::sync::Mutex;
+use async_trait::async_trait;
 use serde_json::{json, Value};
 use std::{
     net::SocketAddr,
@@ -131,8 +132,9 @@ impl ManualSession {
     }*/
 }
 
+#[async_trait]
 impl Session for ManualSession {
-    fn get_info(&self) -> serde_json::Value {
+    async fn get_info(&self) -> serde_json::Value {
         json!({
             "key": 110
         })

src/net/session/outbound_session.rs

@@ -1,6 +1,7 @@
 use async_executor::Executor;
 use async_std::{sync::Mutex, task::yield_now};
-use log::*;
+use async_trait::async_trait;
+use log::{error, info};
 use serde_json::{json, Value};
 use std::{
     net::SocketAddr,
@@ -11,21 +12,76 @@ use crate::{
     error::{Error, Result},
     net::{
         session::{Session, SessionBitflag, SESSION_OUTBOUND},
-        Connector, P2p,
+        ChannelPtr, Connector, P2p,
     },
     system::{StoppableTask, StoppableTaskPtr},
 };
 
+#[derive(Clone)]
+enum OutboundState {
+    Open,
+    Pending,
+    Connected,
+}
+
+impl OutboundState {
+    fn to_string(&self) -> String {
+        match self {
+            Self::Open => "open".to_string(),
+            Self::Pending => "pending".to_string(),
+            Self::Connected => "connected".to_string(),
+        }
+    }
+}
+
+#[derive(Clone)]
+struct OutboundInfo {
+    addr: Option<SocketAddr>,
+    channel: Option<ChannelPtr>,
+    state: OutboundState,
+}
+
+impl OutboundInfo {
+    async fn get_info(&self) -> serde_json::Value {
+        let addr = match self.addr {
+            Some(addr) => serde_json::Value::String(addr.to_string()),
+            None => serde_json::Value::Null,
+        };
+
+        let channel = match &self.channel {
+            Some(channel) => channel.get_info().await,
+            None => serde_json::Value::Null,
+        };
+
+        json!({
+            "addr": addr,
+            "state": self.state.to_string(),
+            "channel": channel,
+        })
+    }
+}
+
+impl Default for OutboundInfo {
+    fn default() -> Self {
+        Self { addr: None, channel: None, state: OutboundState::Open }
+    }
+}
+
 /// Defines outbound connections session.
 pub struct OutboundSession {
     p2p: Weak<P2p>,
     connect_slots: Mutex<Vec<StoppableTaskPtr>>,
+    slot_info: Mutex<Vec<OutboundInfo>>,
 }
 
 impl OutboundSession {
     /// Create a new outbound session.
     pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
-        Arc::new(Self { p2p, connect_slots: Mutex::new(Vec::new()) })
+        Arc::new(Self {
+            p2p,
+            connect_slots: Mutex::new(Vec::new()),
+            slot_info: Mutex::new(Vec::new()),
+        })
     }
 
     /// Start the outbound session. Runs the channel connect loop.
@@ -35,6 +91,8 @@ impl OutboundSession {
         // Activate mutex lock on connection slots.
         let mut connect_slots = self.connect_slots.lock().await;
 
+        self.slot_info.lock().await.resize(slots_count as usize, Default::default());
+
         for i in 0..slots_count {
             let task = StoppableTask::new();
 
@@ -76,6 +134,11 @@ impl OutboundSession {
         loop {
             let addr = self.load_address(slot_number).await?;
             info!(target: "net", "#{} connecting to outbound [{}]", slot_number, addr);
+            {
+                let info = &mut self.slot_info.lock().await[slot_number as usize];
+                info.addr = Some(addr);
+                info.state = OutboundState::Pending;
+            }
 
             match connector.connect(addr).await {
                 Ok(channel) => {
@@ -91,6 +154,12 @@ impl OutboundSession {
 
                     // Remove pending lock since register_channel will add the channel to p2p
                     self.p2p().remove_pending(&addr).await;
+                    {
+                        let info = &mut self.slot_info.lock().await[slot_number as usize];
+                        info.addr = None;
+                        info.channel = Some(channel.clone());
+                        info.state = OutboundState::Connected;
+                    }
 
                     //self.clone().attach_protocols(channel, executor.clone()).await?;
 
@@ -99,6 +168,12 @@ impl OutboundSession {
                 }
                 Err(err) => {
                     info!(target: "net", "Unable to connect to outbound [{}]: {}", addr, err);
+                    {
+                        let info = &mut self.slot_info.lock().await[slot_number as usize];
+                        info.addr = None;
+                        info.channel = None;
+                        info.state = OutboundState::Open;
+                    }
                 }
             }
         }
@@ -170,10 +245,17 @@ impl OutboundSession {
     }*/
 }
 
+#[async_trait]
 impl Session for OutboundSession {
-    fn get_info(&self) -> serde_json::Value {
+    async fn get_info(&self) -> serde_json::Value {
+        let mut slots = Vec::new();
+        for info in &*self.slot_info.lock().await {
+            slots.push(info.get_info().await);
+        }
+
         json!({
-            "key": 110
+            "key": 110,
+            "slots": slots,
         })
     }
 

src/net/session/seed_session.rs

@@ -1,4 +1,5 @@
 use async_std::future::timeout;
+use async_trait::async_trait;
 use serde_json::{json, Value};
 use std::{
     net::SocketAddr,
@@ -136,8 +137,9 @@ impl SeedSession {
     }*/
 }
 
+#[async_trait]
 impl Session for SeedSession {
-    fn get_info(&self) -> serde_json::Value {
+    async fn get_info(&self) -> serde_json::Value {
         json!({
             "key": 110
         })

src/net/session/session.rs

@@ -96,7 +96,7 @@ pub trait Session: Sync {
         Ok(())
     }
 
-    fn get_info(&self) -> serde_json::Value;
+    async fn get_info(&self) -> serde_json::Value;
 
     /// Returns a pointer to the p2p network interface.
     fn p2p(&self) -> P2pPtr;