Enforce client semantics with state

Uses the sealed trait hack to only allow for compiler enforced state
transactions. There is first a `run` step, followed by the typical
methods a client would expect. I decided to scrap the `Requester` and
instead encapsulate the event receivers with a newtype. Running a client
gives access to the event receivers as well as the handle to the task.

The only downside I see with this is `&mut self` doesn't seem to be
possible, as each impl block is defined for it's `State` type. I need to
fiddle with that to see if we can't remove the need to return the client
each state transition.

Author: rustaceanrob

examples/bitcoin.rs

@@ -3,7 +3,8 @@
 
 use bip157::builder::Builder;
 use bip157::chain::{BlockHeaderChanges, ChainState};
-use bip157::{lookup_host, Client, Event, HeaderCheckpoint, Network, ScriptBuf};
+use bip157::client::EventListeners;
+use bip157::{lookup_host, Event, HeaderCheckpoint, Network, ScriptBuf};
 use std::collections::HashSet;
 use tokio::time::Instant;
 
@@ -34,18 +35,13 @@ async fn main() {
         .add_peers(seeds.into_iter().map(From::from))
         // Create the node and client
         .build();
-
-    let (client, _) = client.run();
-    // Split the client into components that send messages and listen to messages.
-    // With this construction, different parts of the program can take ownership of
-    // specific tasks.
-    let Client {
-        requester,
+    let (client, events, _) = client.run();
+    let EventListeners {
         mut info_rx,
         mut warn_rx,
         mut event_rx,
-        ..
-    } = client;
+    } = events;
+
     // Continually listen for events until the node is synced to its peers.
     loop {
         tokio::select! {
@@ -55,11 +51,11 @@ async fn main() {
                         Event::FiltersSynced(update) => {
                             tracing::info!("Chain tip: {}",update.tip().hash);
                             // Request information from the node
-                            let fee = requester.broadcast_min_feerate().await.unwrap();
+                            let fee = client.broadcast_min_feerate().await.unwrap();
                             tracing::info!("Minimum transaction broadcast fee rate: {:#}", fee);
                             let sync_time = now.elapsed().as_secs_f32();
                             tracing::info!("Total sync time: {sync_time} seconds");
-                            let avg_fee_rate = requester.average_fee_rate(update.tip().hash).await.unwrap();
+                            let avg_fee_rate = client.average_fee_rate(update.tip().hash).await.unwrap();
                             tracing::info!("Last block average fee rate: {:#}", avg_fee_rate);
                             break;
                         },
@@ -82,6 +78,6 @@ async fn main() {
             }
         }
     }
-    let _ = requester.shutdown();
+    let _ = client.shutdown();
     tracing::info!("Shutting down");
 }

examples/signet.rs

@@ -3,8 +3,8 @@
 
 use bip157::chain::{BlockHeaderChanges, ChainState};
 use bip157::messages::Event;
-use bip157::{builder::Builder, chain::checkpoints::HeaderCheckpoint, Client};
-use bip157::{Address, BlockHash, Network};
+use bip157::{builder::Builder, chain::checkpoints::HeaderCheckpoint};
+use bip157::{Address, BlockHash, EventListeners, Network};
 use std::collections::HashSet;
 use std::str::FromStr;
 
@@ -40,15 +40,12 @@ async fn main() {
         // Create the node and client
         .build();
 
-    let (client, _) = client.run();
-
-    let Client {
-        requester,
+    let (client, events, _) = client.run();
+    let EventListeners {
         mut info_rx,
         mut warn_rx,
         mut event_rx,
-        ..
-    } = client;
+    } = events;
 
     // Continually listen for events until the node is synced to its peers.
     loop {
@@ -72,7 +69,7 @@ async fn main() {
                             if filter.contains_any(addresses.iter()) {
                                 let hash = filter.block_hash();
                                 tracing::info!("Found script at {}!", hash);
-                                let indexed_block = requester.get_block(hash).await.unwrap();
+                                let indexed_block = client.get_block(hash).await.unwrap();
                                 let coinbase = indexed_block.block.txdata.first().unwrap().compute_txid();
                                 tracing::info!("Coinbase transaction ID: {}", coinbase);
                                 break;
@@ -88,6 +85,6 @@ async fn main() {
             }
         }
     }
-    let _ = requester.shutdown();
+    let _ = client.shutdown();
     tracing::info!("Shutting down");
 }

src/builder.rs

@@ -5,6 +5,7 @@ use bitcoin::Network;
 
 use super::{client::Client, node::Node};
 use crate::chain::ChainState;
+use crate::client::Idle;
 use crate::network::ConnectionType;
 use crate::TrustedPeer;
 use crate::{Config, FilterType};
@@ -139,7 +140,7 @@ impl Builder {
     }
 
     /// Consume the node builder and receive a [`Client`].
-    pub fn build(mut self) -> Client {
+    pub fn build(mut self) -> Client<Idle> {
         Node::build(self.network, core::mem::take(&mut self.config))
     }
 }

src/client.rs

@@ -13,50 +13,106 @@ use crate::{Event, Info, TrustedPeer, Warning};
 use super::{error::ClientError, messages::ClientMessage};
 use super::{error::FetchBlockError, IndexedBlock};
 
-/// A [`Client`] allows for communication with a running node.
+/// Client state when idle.
+pub struct Idle;
+/// Client state when active.
+pub struct Active;
+
+mod sealed {
+    pub trait Sealed {}
+}
+
+impl sealed::Sealed for Idle {}
+impl sealed::Sealed for Active {}
+
+/// State of the client
+pub trait State: sealed::Sealed {}
+
+impl State for Idle {}
+impl State for Active {}
+
+/// Wrapper type for the channels that will receive events.
 #[derive(Debug)]
-pub struct Client {
-    /// Send events to a node, such as broadcasting a transaction.
-    pub requester: Requester,
+pub struct EventListeners {
     /// Receive informational messages from the node.
     pub info_rx: mpsc::Receiver<Info>,
     /// 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>,
-    /// Internal node structure.
-    node: Option<Node>,
 }
 
-impl Client {
-    pub(crate) fn new(
+impl EventListeners {
+    fn new(
         info_rx: mpsc::Receiver<Info>,
         warn_rx: mpsc::UnboundedReceiver<Warning>,
         event_rx: mpsc::UnboundedReceiver<Event>,
-        ntx: UnboundedSender<ClientMessage>,
-        node: Node,
     ) -> Self {
         Self {
-            requester: Requester::new(ntx),
             info_rx,
             warn_rx,
             event_rx,
-            node: Some(node),
         }
     }
+}
+
+/// A [`Client`] allows for communication with a running node.
+#[derive(Debug)]
+pub struct Client<S: State> {
+    /// Send events to a node, such as broadcasting a transaction.
+    ntx: UnboundedSender<ClientMessage>,
+    /// Receive informational messages from the node.
+    events: Option<EventListeners>,
+    /// Internal node structure.
+    node: Option<Node>,
+    /// Marker for state.
+    _marker: core::marker::PhantomData<S>,
+}
 
+impl Client<Idle> {
+    pub(crate) fn new(
+        info_rx: mpsc::Receiver<Info>,
+        warn_rx: mpsc::UnboundedReceiver<Warning>,
+        event_rx: mpsc::UnboundedReceiver<Event>,
+        ntx: UnboundedSender<ClientMessage>,
+        node: Node,
+    ) -> Client<Idle> {
+        Client {
+            ntx,
+            events: Some(EventListeners::new(info_rx, warn_rx, event_rx)),
+            node: Some(node),
+            _marker: core::marker::PhantomData,
+        }
+    }
     /// Start the underlying node on a [`tokio::task`]. This assumes there is a runtime present to
     /// execute the task.
-    pub fn run(mut self) -> (Self, JoinHandle<Result<(), crate::error::NodeError>>) {
+    pub fn run(
+        mut self,
+    ) -> (
+        Client<Active>,
+        EventListeners,
+        JoinHandle<Result<(), crate::error::NodeError>>,
+    ) {
+        let events = core::mem::take(&mut self.events).expect("cannot call run twice.");
         let node = core::mem::take(&mut self.node).expect("cannot call run twice.");
-        (self, tokio::task::spawn(async move { node.run().await }))
+        (
+            Client {
+                ntx: self.ntx,
+                events: None,
+                node: None,
+                _marker: core::marker::PhantomData,
+            },
+            events,
+            tokio::task::spawn(async move { node.run().await }),
+        )
     }
 
     /// Run on a detached operating system thread. This method is useful in the case where the
     /// majority of your application code is blocking, and you do not have a
     /// [`tokio::runtime::Runtime`] available. This method will implicitly create a runtime which
     /// runs the data fetching process.
-    pub fn run_detached(mut self) -> (Self, std::thread::JoinHandle<()>) {
+    pub fn run_detached(mut self) -> (Client<Active>, EventListeners, std::thread::JoinHandle<()>) {
+        let events = core::mem::take(&mut self.events).expect("cannot call run twice.");
         let node = core::mem::take(&mut self.node).expect("cannot call run twice.");
         let handle = std::thread::spawn(|| {
             tokio::runtime::Builder::new_multi_thread()
@@ -67,27 +123,37 @@ impl Client {
                     let _ = node.run().await;
                 })
         });
-        (self, handle)
+        let client = Client {
+            ntx: self.ntx,
+            events: None,
+            node: None,
+            _marker: core::marker::PhantomData,
+        };
+        (client, events, handle)
     }
 
     /// Run the node with an existing [`tokio::runtime::Runtime`].
-    pub fn run_with_runtime(mut self, rt: tokio::runtime::Runtime) -> (Self, JoinHandle<Result<(), crate::error::NodeError>>) {
+    pub fn run_with_runtime(
+        mut self,
+        rt: tokio::runtime::Runtime,
+    ) -> (
+        Client<Active>,
+        EventListeners,
+        JoinHandle<Result<(), crate::error::NodeError>>,
+    ) {
+        let events = core::mem::take(&mut self.events).expect("cannot call run twice.");
         let node = core::mem::take(&mut self.node).expect("cannot call run twice.");
-        (self, rt.spawn(async move { node.run().await }))
+        let client = Client {
+            ntx: self.ntx,
+            events: None,
+            node: None,
+            _marker: core::marker::PhantomData,
+        };
+        (client, events, rt.spawn(async move { node.run().await }))
     }
 }
 
-/// Send messages to a node that is running so the node may complete a task.
-#[derive(Debug, Clone)]
-pub struct Requester {
-    ntx: UnboundedSender<ClientMessage>,
-}
-
-impl Requester {
-    fn new(ntx: UnboundedSender<ClientMessage>) -> Self {
-        Self { ntx }
-    }
-
+impl Client<Active> {
     /// Tell the node to shut down.
     ///
     /// # Errors

src/lib.rs

@@ -8,7 +8,7 @@
 //! # Example usage
 //!
 //! ```no_run
-//! use bip157::{Builder, Event, Client, Network, BlockHash};
+//! use bip157::{Builder, Event, EventListeners, Client, Network, BlockHash};
 //!
 //! #[tokio::main]
 //! async fn main() {
@@ -22,10 +22,11 @@
 //!         // The number of connections we would like to maintain
 //!         .required_peers(2)
 //!         .build();
-//!     // Run the node and wait for the sync message;
-//!     let (client, _) = client.run();
+//!     // Start the node
+//!     let (client, events, _) = client.run();
 //!     // Split the client into components that send messages and listen to messages
-//!     let Client { requester, info_rx: _, warn_rx: _, mut event_rx, .. } = client;
+//!     let EventListeners { info_rx: _, warn_rx: _, mut event_rx } = events;
+//!     // Wait for the sync message;
 //!     loop {
 //!         if let Some(event) = event_rx.recv().await {
 //!             match event {
@@ -37,7 +38,7 @@
 //!             }
 //!         }
 //!     }
-//!     requester.shutdown();
+//!     client.shutdown();
 //! }
 //! ```
 
@@ -81,7 +82,7 @@ use tokio::sync::mpsc::UnboundedSender;
 pub use {
     crate::builder::Builder,
     crate::chain::ChainState,
-    crate::client::{Client, Requester},
+    crate::client::{Client, EventListeners},
     crate::error::{ClientError, NodeError},
     crate::messages::{Event, Info, Progress, RejectPayload, SyncUpdate, Warning},
 };

src/node.rs

@@ -31,6 +31,7 @@ use crate::{
         error::HeaderSyncError,
         CFHeaderChanges, ChainState, FilterCheck, HeightMonitor,
     },
+    client::Idle,
     error::FetchBlockError,
     messages::ClientRequest,
     network::{
@@ -66,7 +67,7 @@ pub(crate) struct Node {
 }
 
 impl Node {
-    pub(crate) fn build(network: Network, config: Config) -> Client {
+    pub(crate) fn build(network: Network, config: Config) -> Client<Idle> {
         let Config {
             required_peers,
             white_list,