feat(client)!: split Warning into separate receiver

Author: rustaceanrob

example/managed.rs

@@ -54,6 +54,7 @@ async fn main() {
     let Client {
         requester,
         mut log_rx,
+        warn_rx: _,
         mut event_rx,
     } = client;
 
@@ -62,10 +63,8 @@ async fn main() {
         tokio::select! {
             log = log_rx.recv() => {
                 if let Some(log) = log {
-                    match log {
-                        Log::Dialog(d) => tracing::info!("{d}"),
-                        Log::Warning(e) => tracing::warn!("{e}"),
-                        _ => (),
+                    if let Log::Dialog(log) = log {
+                        tracing::info!("{log}")
                     }
                 }
             }

example/rescan.rs

@@ -3,7 +3,7 @@
 //! blocks.
 
 use kyoto::{chain::checkpoints::HeaderCheckpoint, core::builder::NodeBuilder};
-use kyoto::{Address, Client, Event, Log, Network};
+use kyoto::{Address, Client, Event, Network};
 use std::collections::HashSet;
 use std::str::FromStr;
 
@@ -45,6 +45,7 @@ async fn main() {
     let Client {
         requester,
         mut log_rx,
+        mut warn_rx,
         mut event_rx,
     } = client;
     // Sync with the single script added
@@ -59,11 +60,12 @@ async fn main() {
             }
             log = log_rx.recv() => {
                 if let Some(log) = log {
-                    match log {
-                        Log::Dialog(d) => tracing::info!("{d}"),
-                        Log::Warning(warning) => tracing::warn!("{warning}"),
-                        _ => (),
-                    }
+                    tracing::info!("{log}");
+                }
+            }
+            warn = warn_rx.recv() => {
+                if let Some(warn) = warn {
+                    tracing::warn!("{warn}");
                 }
             }
         }
@@ -89,11 +91,12 @@ async fn main() {
             }
             log = log_rx.recv() => {
                 if let Some(log) = log {
-                    match log {
-                        Log::Dialog(d) => tracing::info!("{d}"),
-                        Log::Warning(warning) => tracing::warn!("{warning}"),
-                        _ => (),
-                    }
+                    tracing::info!("{log}");
+                }
+            }
+            warn = warn_rx.recv() => {
+                if let Some(warn) = warn {
+                    tracing::warn!("{warn}");
                 }
             }
         }

example/signet.rs

@@ -57,6 +57,7 @@ async fn main() {
     let Client {
         requester,
         mut log_rx,
+        mut warn_rx,
         mut event_rx,
     } = client;
     // Continually listen for events until the node is synced to its peers.
@@ -87,13 +88,17 @@ async fn main() {
                 if let Some(log) = log {
                     match log {
                         Log::Dialog(d)=> tracing::info!("{d}"),
-                        Log::Warning(warning)=> tracing::warn!("{warning}"),
                         Log::StateChange(node_state) => tracing::info!("{node_state}"),
                         Log::ConnectionsMet => tracing::info!("All required connections met"),
                         _ => (),
                     }
                 }
             }
+            warn = warn_rx.recv() => {
+                if let Some(warn) = warn {
+                    tracing::warn!("{warn}");
+                }
+            }
         }
     }
     let _ = requester.shutdown().await;

example/testnet4.rs

@@ -52,6 +52,7 @@ async fn main() {
     let Client {
         requester,
         mut log_rx,
+        mut warn_rx,
         mut event_rx,
     } = client;
     // Continually listen for events until the node is synced to its peers.
@@ -79,13 +80,17 @@ async fn main() {
                 if let Some(log) = log {
                     match log {
                         Log::Dialog(d)=> tracing::info!("{d}"),
-                        Log::Warning(warning)=> tracing::warn!("{warning}"),
                         Log::StateChange(node_state) => tracing::info!("{node_state}"),
                         Log::ConnectionsMet => tracing::info!("All required connections met"),
                         _ => (),
                     }
                 }
             }
+            warn = warn_rx.recv() => {
+                if let Some(warn) = warn {
+                    tracing::warn!("{warn}");
+                }
+            }
         }
     }
     let _ = requester.shutdown().await;

src/chain/chain.rs

@@ -949,7 +949,7 @@ mod tests {
         },
         core::{
             dialog::Dialog,
-            messages::{Event, Log},
+            messages::{Event, Log, Warning},
         },
         filters::cfheader_chain::AppendAttempt,
     };
@@ -958,6 +958,7 @@ mod tests {
 
     fn new_regtest(anchor: HeaderCheckpoint) -> Chain<()> {
         let (log_tx, _) = tokio::sync::mpsc::channel::<Log>(1);
+        let (warn_tx, _) = tokio::sync::mpsc::unbounded_channel::<Warning>();
         let (event_tx, _) = tokio::sync::mpsc::unbounded_channel::<Event>();
         let mut checkpoints = HeaderCheckpoints::new(&bitcoin::Network::Regtest);
         checkpoints.prune_up_to(anchor);
@@ -966,14 +967,15 @@ mod tests {
             HashSet::new(),
             anchor,
             checkpoints,
-            Dialog::new(log_tx, event_tx),
+            Dialog::new(log_tx, warn_tx, event_tx),
             (),
             1,
         )
     }
 
     fn new_regtest_two_peers(anchor: HeaderCheckpoint) -> Chain<()> {
         let (log_tx, _) = tokio::sync::mpsc::channel::<Log>(1);
+        let (warn_tx, _) = tokio::sync::mpsc::unbounded_channel::<Warning>();
         let (event_tx, _) = tokio::sync::mpsc::unbounded_channel::<Event>();
         let mut checkpoints = HeaderCheckpoints::new(&bitcoin::Network::Regtest);
         checkpoints.prune_up_to(anchor);
@@ -982,7 +984,7 @@ mod tests {
             HashSet::new(),
             anchor,
             checkpoints,
-            Dialog::new(log_tx, event_tx),
+            Dialog::new(log_tx, warn_tx, event_tx),
             (),
             2,
         )

src/core/client.rs

@@ -8,7 +8,7 @@ use std::{collections::BTreeMap, ops::Range, time::Duration};
 use tokio::sync::mpsc;
 use tokio::sync::mpsc::Sender;
 
-use crate::{Event, Log, TrustedPeer, TxBroadcast};
+use crate::{Event, Log, TrustedPeer, TxBroadcast, Warning};
 
 #[cfg(feature = "filter-control")]
 use super::{error::FetchBlockError, messages::BlockRequest, BlockReceiver, IndexedBlock};
@@ -24,19 +24,23 @@ pub struct Client {
     pub requester: Requester,
     /// Receive log messages from a node.
     pub log_rx: mpsc::Receiver<Log>,
+    /// Receive warning messages from a node.
+    pub warn_rx: mpsc::UnboundedReceiver<Warning>,
     /// Receive [`Event`] from a node to act on.
     pub event_rx: mpsc::UnboundedReceiver<Event>,
 }
 
 impl Client {
     pub(crate) fn new(
         log_rx: mpsc::Receiver<Log>,
+        warn_rx: mpsc::UnboundedReceiver<Warning>,
         event_rx: mpsc::UnboundedReceiver<Event>,
         ntx: Sender<ClientMessage>,
     ) -> Self {
         Self {
             requester: Requester::new(ntx),
             log_rx,
+            warn_rx,
             event_rx,
         }
     }
@@ -351,13 +355,15 @@ mod tests {
     async fn test_client_works() {
         let transaction: Transaction = deserialize(&hex::decode("0200000001aad73931018bd25f84ae400b68848be09db706eac2ac18298babee71ab656f8b0000000048473044022058f6fc7c6a33e1b31548d481c826c015bd30135aad42cd67790dab66d2ad243b02204a1ced2604c6735b6393e5b41691dd78b00f0c5942fb9f751856faa938157dba01feffffff0280f0fa020000000017a9140fb9463421696b82c833af241c78c17ddbde493487d0f20a270100000017a91429ca74f8a08f81999428185c97b5d852e4063f618765000000").unwrap()).unwrap();
         let (log_tx, log_rx) = tokio::sync::mpsc::channel::<Log>(1);
+        let (_, warn_rx) = tokio::sync::mpsc::unbounded_channel::<Warning>();
         let (_, event_rx) = tokio::sync::mpsc::unbounded_channel::<Event>();
         let (ctx, crx) = mpsc::channel::<ClientMessage>(5);
         let Client {
             requester,
             mut log_rx,
+            warn_rx: _,
             event_rx: _,
-        } = Client::new(log_rx, event_rx, ctx);
+        } = Client::new(log_rx, warn_rx, event_rx, ctx);
         let send_res = log_tx
             .send(Log::Dialog("An important message".into()))
             .await;

src/core/dialog.rs

@@ -5,12 +5,21 @@ use super::messages::{Event, Log, Progress, Warning};
 #[derive(Debug, Clone)]
 pub(crate) struct Dialog {
     log_tx: Sender<Log>,
+    warn_tx: UnboundedSender<Warning>,
     event_tx: UnboundedSender<Event>,
 }
 
 impl Dialog {
-    pub(crate) fn new(log_tx: Sender<Log>, event_tx: UnboundedSender<Event>) -> Self {
-        Self { log_tx, event_tx }
+    pub(crate) fn new(
+        log_tx: Sender<Log>,
+        warn_tx: UnboundedSender<Warning>,
+        event_tx: UnboundedSender<Event>,
+    ) -> Self {
+        Self {
+            log_tx,
+            warn_tx,
+            event_tx,
+        }
     }
 
     pub(crate) async fn send_dialog(&self, dialog: impl Into<String>) {
@@ -40,7 +49,7 @@ impl Dialog {
     }
 
     pub(crate) async fn send_warning(&self, warning: Warning) {
-        let _ = self.log_tx.send(Log::Warning(warning)).await;
+        let _ = self.warn_tx.send(warning);
     }
 
     pub(crate) async fn send_info(&self, info: Log) {

src/core/messages.rs

@@ -21,8 +21,6 @@ use super::{
 pub enum Log {
     /// Human readable dialog of what the node is currently doing.
     Dialog(String),
-    /// A warning that may effect the function of the node.
-    Warning(Warning),
     /// The current state of the node in the syncing process.
     StateChange(NodeState),
     /// The node is connected to all required peers.
@@ -39,7 +37,6 @@ impl core::fmt::Display for Log {
     fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
         match self {
             Log::Dialog(d) => write!(f, "{}", d),
-            Log::Warning(w) => write!(f, "{}", w),
             Log::StateChange(s) => write!(f, "{}", s),
             Log::TxSent(txid) => write!(f, "Transaction sent: {}", txid),
             Log::ConnectionsMet => write!(f, "Required connections met"),

src/core/node.rs

@@ -93,11 +93,12 @@ impl<H: HeaderStore, P: PeerStore> Node<H, P> {
     ) -> (Self, Client) {
         // Set up a communication channel between the node and client
         let (log_tx, log_rx) = mpsc::channel::<Log>(32);
+        let (warn_tx, warn_rx) = mpsc::unbounded_channel::<Warning>();
         let (event_tx, event_rx) = mpsc::unbounded_channel::<Event>();
         let (ctx, crx) = mpsc::channel::<ClientMessage>(5);
-        let client = Client::new(log_rx, event_rx, ctx);
+        let client = Client::new(log_rx, warn_rx, event_rx, ctx);
         // A structured way to talk to the client
-        let dialog = Dialog::new(log_tx, event_tx);
+        let dialog = Dialog::new(log_tx, warn_tx, event_tx);
         // We always assume we are behind
         let state = Arc::new(RwLock::new(NodeState::Behind));
         // Configure the peer manager

src/lib.rs

@@ -42,7 +42,7 @@
 //!     // Run the node and wait for the sync message;
 //!     tokio::task::spawn(async move { node.run().await });
 //!     // Split the client into components that send messages and listen to messages
-//!     let Client { requester, mut log_rx, mut event_rx } = client;
+//!     let Client { requester, mut log_rx, warn_rx: _, mut event_rx } = client;
 //!     // Sync with the single script added
 //!     loop {
 //!         tokio::select! {