ran cargo fmt w new defaults

src/net/acceptor.rs

@@ -10,8 +10,9 @@ use crate::system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr,
 /// Atomic pointer to Acceptor class.
 pub type AcceptorPtr = Arc<Acceptor>;
 
-/// Handles the acceptance of inbound socket connections. Used to start listening
-/// on a local socket address, to accept incoming connections and to handle network errors.
+/// Handles the acceptance of inbound socket connections. Used to start
+/// listening on a local socket address, to accept incoming connections and to
+/// handle network errors.
 pub struct Acceptor {
     channel_subscriber: SubscriberPtr<NetResult<ChannelPtr>>,
     task: StoppableTaskPtr,
@@ -73,7 +74,8 @@ impl Acceptor {
         Ok(listener)
     }
 
-    /// Run the accept loop in a new thread and error if a connection problem occurs.
+    /// Run the accept loop in a new thread and error if a connection problem
+    /// occurs.
     fn accept(self: Arc<Self>, listener: Async<TcpListener>, executor: Arc<Executor<'_>>) {
         self.task.clone().start(
             self.clone().run_accept_loop(listener),
@@ -91,7 +93,8 @@ impl Acceptor {
         }
     }
 
-    /// Handles network errors. Panics if error passes silently, otherwise broadcasts the error.
+    /// Handles network errors. Panics if error passes silently, otherwise
+    /// broadcasts the error.
     async fn handle_stop(self: Arc<Self>, result: NetResult<()>) {
         match result {
             Ok(()) => panic!("Acceptor task should never complete without error status"),
@@ -103,7 +106,8 @@ impl Acceptor {
         }
     }
 
-    /// Single attempt to accept an incoming connection. Stops after one attempt.
+    /// Single attempt to accept an incoming connection. Stops after one
+    /// attempt.
     async fn tick_accept(&self, listener: &Async<TcpListener>) -> NetResult<ChannelPtr> {
         let (stream, peer_addr) = match listener.accept().await {
             Ok((s, a)) => (s, a),

src/net/channel.rs

@@ -18,7 +18,10 @@ use crate::system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr,
 /// Atomic pointer to async channel.
 pub type ChannelPtr = Arc<Channel>;
 
-/// Async TCP channel that handles the sending of messages across the network.
+/// Async channel interface that handles the sending of messages across the
+/// network. Public interface is used to create new channels, to stop and start
+/// a channel, send messages. Also implements message functionality. Implements
+/// the message subscriber subsystem.
 pub struct Channel {
     reader: Mutex<ReadHalf<Async<TcpStream>>>,
     writer: Mutex<WriteHalf<Async<TcpStream>>>,
@@ -185,11 +188,13 @@ impl Channel {
             .await;
     }
 
+    /// Convenience function that returns the Message Subsystem.
     pub fn get_message_subsystem(&self) -> &MessageSubsystem {
         &self.message_subsystem
     }
 
-    /// Run the receive loop. Start receiving messages or handle network failure.
+    /// Run the receive loop. Start receiving messages or handle network
+    /// failure.
     async fn main_receive_loop(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net",
             "Channel::receive_loop() [START, address={}]",
@@ -223,7 +228,8 @@ impl Channel {
         }
     }
 
-    /// Handle network errors. Panic if error passes silently, otherwise broadcast the error.
+    /// Handle network errors. Panic if error passes silently, otherwise
+    /// broadcast the error.
     async fn handle_stop(self: Arc<Self>, result: NetResult<()>) {
         debug!(target: "net", "Channel::handle_stop() [START, address={}]", self.address());
         match result {

src/net/message_subscriber.rs

@@ -53,7 +53,8 @@ trait MessageDispatcherInterface: Send + Sync {
     fn as_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync>;
 }
 
-/// Maintains a list of active subscribers and handles sending messages across subscriptions.
+/// Maintains a list of active subscribers and handles sending messages across
+/// subscriptions.
 struct MessageDispatcher<M: Message> {
     subs: Mutex<HashMap<MessageSubscriptionID, async_channel::Sender<MessageResult<M>>>>,
 }
@@ -72,7 +73,8 @@ impl<M: Message> MessageDispatcher<M> {
         rng.gen()
     }
 
-    /// Subscribe to a channel. Assigns a new ID and adds it to the list of subscribers.
+    /// Subscribe to a channel. Assigns a new ID and adds it to the list of
+    /// subscribers.
     pub async fn subscribe(self: Arc<Self>) -> MessageSubscription<M> {
         let (sender, recvr) = async_channel::unbounded();
         let sub_id = Self::random_id();
@@ -85,12 +87,14 @@ impl<M: Message> MessageDispatcher<M> {
         }
     }
 
-    /// Unsubcribe from a channel. Removes the associated ID from the subscriber list.
+    /// Unsubcribe from a channel. Removes the associated ID from the subscriber
+    /// list.
     async fn unsubscribe(&self, sub_id: MessageSubscriptionID) {
         self.subs.lock().await.remove(&sub_id);
     }
 
-    /// Send a message to all subscriber channels. Automatically clear inactive channels.
+    /// Send a message to all subscriber channels. Automatically clear inactive
+    /// channels.
     async fn trigger_all(&self, message: MessageResult<M>) {
         debug!(
             "MessageDispatcher<M={}>::trigger_all({}) [START, subs={}]",
@@ -106,7 +110,8 @@ impl<M: Message> MessageDispatcher<M> {
                 Err(_err) => {
                     // Automatically clean out closed channels
                     garbage_ids.push(*sub_id);
-                    //panic!("Error returned sending message in notify() call! {}", err);
+                    // panic!("Error returned sending message in notify() call!
+                    // {}", err);
                 }
             }
         }
@@ -149,7 +154,8 @@ impl<M: Message> MessageDispatcherInterface for MessageDispatcher<M> {
         }
     }
 
-    /// Sends a message to all subscriber channels. Clears any inactive channels.
+    /// Sends a message to all subscriber channels. Clears any inactive
+    /// channels.
     async fn trigger_error(&self, err: NetError) {
         self.trigger_all(Err(err)).await;
     }
@@ -160,10 +166,22 @@ impl<M: Message> MessageDispatcherInterface for MessageDispatcher<M> {
     }
 }
 
-/// Generic publish/subscribe class that can dispatch any kind of message
-/// to a subscribed list of dispatchers. Dispatchers subscribe to a single
-/// message format of any type. This is a generalized version of the pub/sub
-/// model in system::Subscriber.
+/// Generic publish/subscribe class that can dispatch any kind of message to a
+/// subscribed list of dispatchers. Dispatchers subscribe to a single
+/// message format of any type. This is a generalized version of the simple
+/// publish-subscribe class in system::Subscriber.
+///
+/// Message Subsystem maintains a list of dispatchers. Dispatchers belong to
+/// a class of subscribers called Message Dispatcher. Message Dispatcher
+/// implements a generic trait called Message Dispatcher Interface.
+///
+/// Pub-sub is called on dispatchers through the functions 'subscribe' and
+/// 'notify'. Whereas system::Subscriber only allows messages of a single type,
+/// dispatchers can handle any kind of message. This generic message is called a
+/// a payload and is processed and decoded by the Message Dispatcher.
+///
+/// Message Subsystem also enables the creation of new message subsystems,
+/// adding new dispatchers and clearing inactive channels.
 pub struct MessageSubsystem {
     dispatchers: Mutex<HashMap<&'static str, Arc<dyn MessageDispatcherInterface>>>,
 }
@@ -207,7 +225,8 @@ impl MessageSubsystem {
         Ok(sub)
     }
 
-    /// Sends a message out to subscribers. Returns an error if the message doesn't send.
+    /// Sends a message out to subscribers. Returns an error if the message
+    /// doesn't send.
     pub async fn notify(&self, command: &str, payload: Vec<u8>) {
         let dispatcher = self.dispatchers.lock().await.get(command).cloned();
 
@@ -236,7 +255,8 @@ impl MessageSubsystem {
 /// Test functions for message subsystem.
 // This is a test function for the message subsystem code above
 // Normall we would use the #[test] macro but cannot since it is async code
-// Instead we call it using smol::block_on() in the unit test code after this func
+// Instead we call it using smol::block_on() in the unit test code after this
+// func
 async fn _do_message_subscriber_test() {
     struct MyVersionMessage {
         x: u32,

src/net/messages.rs

@@ -26,7 +26,8 @@ pub struct PongMessage {
 /// Requests address of outbound connection.
 pub struct GetAddrsMessage {}
 
-/// Sends address information to inbound connection. Response to GetAddrs message.
+/// Sends address information to inbound connection. Response to GetAddrs
+/// message.
 pub struct AddrsMessage {
     pub addrs: Vec<SocketAddr>,
 }

src/net/protocols/protocol_address.rs

@@ -21,8 +21,8 @@ pub struct ProtocolAddress {
 }
 
 impl ProtocolAddress {
-    /// Create a new address protocol. Makes an address and get-address subscription
-    /// and adds them to the address protocol instance.
+    /// Create a new address protocol. Makes an address and get-address
+    /// subscription and adds them to the address protocol instance.
     pub async fn new(channel: ChannelPtr, hosts: HostsPtr) -> Arc<Self> {
         // Creates a subscription to address message.
         let addrs_sub = channel
@@ -47,8 +47,9 @@ impl ProtocolAddress {
         })
     }
 
-    /// Starts the address protocol. Runs receive address and get address protocols
-    /// on the protocol task manager. Then sends get-address message.
+    /// Starts the address protocol. Runs receive address and get address
+    /// protocols on the protocol task manager. Then sends get-address
+    /// message.
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
         debug!(target: "net", "ProtocolAddress::start() [START]");
         self.jobsman.clone().start(executor.clone());
@@ -67,8 +68,9 @@ impl ProtocolAddress {
         debug!(target: "net", "ProtocolAddress::start() [END]");
     }
 
-    /// Handles receiving the address message. Loops to continually recieve address
-    /// messages on the address subsciption. Adds the recieved addresses to the list of hosts.
+    /// Handles receiving the address message. Loops to continually recieve
+    /// address messages on the address subsciption. Adds the recieved
+    /// addresses to the list of hosts.
     async fn handle_receive_addrs(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net", "ProtocolAddress::handle_receive_addrs() [START]");
         loop {
@@ -86,8 +88,9 @@ impl ProtocolAddress {
         }
     }
 
-    /// Handles receiving the get-address message. Continually recieves get-address
-    /// messages on the get-address subsciption. Then replies with an address message.
+    /// Handles receiving the get-address message. Continually recieves
+    /// get-address messages on the get-address subsciption. Then replies
+    /// with an address message.
     async fn handle_receive_get_addrs(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() [START]");
         loop {

src/net/protocols/protocol_jobs_manager.rs

@@ -28,7 +28,8 @@ impl ProtocolJobsManager {
         })
     }
 
-    /// Runs the task on an executor. Prepares to stop all tasks when the channel is closed.
+    /// Runs the task on an executor. Prepares to stop all tasks when the
+    /// channel is closed.
     pub fn start(self: Arc<Self>, executor: ExecutorPtr<'_>) {
         executor.spawn(self.handle_stop()).detach()
     }
@@ -41,8 +42,8 @@ impl ProtocolJobsManager {
         self.tasks.lock().await.push(executor.spawn(future))
     }
 
-    /// Waits for a stop signal, then closes all tasks. Insures that all tasks are
-    /// stopped when a channel closes. Called in start().
+    /// Waits for a stop signal, then closes all tasks. Insures that all tasks
+    /// are stopped when a channel closes. Called in start().
     async fn handle_stop(self: Arc<Self>) {
         let stop_sub = self.channel.clone().subscribe_stop().await;
 
@@ -53,7 +54,8 @@ impl ProtocolJobsManager {
         self.close_all_tasks().await
     }
 
-    /// Closes all open tasks. Takes all the tasks from the internal queue and closes them.
+    /// Closes all open tasks. Takes all the tasks from the internal queue and
+    /// closes them.
     async fn close_all_tasks(self: Arc<Self>) {
         debug!(target: "net",
             "ProtocolJobsManager::close_all_tasks() [START, name={}, addr={}]",

src/net/protocols/protocol_ping.rs

@@ -28,9 +28,9 @@ impl ProtocolPing {
         })
     }
 
-    /// Starts ping-pong keep-alive messages exchange. Runs ping-pong in the protocol
-    /// task manager, then queues the reply. Sends out a ping and waits for pong
-    /// reply. Waits for ping and replies with a pong.
+    /// Starts ping-pong keep-alive messages exchange. Runs ping-pong in the
+    /// protocol task manager, then queues the reply. Sends out a ping and
+    /// waits for pong reply. Waits for ping and replies with a pong.
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
         debug!(target: "net", "ProtocolPing::start() [START]");
         self.jobsman.clone().start(executor.clone());
@@ -45,10 +45,10 @@ impl ProtocolPing {
         debug!(target: "net", "ProtocolPing::start() [END]");
     }
 
-    /// Runs ping-pong protocol. Creates a subscription to pong, then starts a loop.
-    /// Loop sleeps for the duration of the channel heartbeat, then sends a ping
-    /// message with a random nonce. Loop starts a timer, waits for the pong reply
-    /// and insures the nonce is the same.
+    /// Runs ping-pong protocol. Creates a subscription to pong, then starts a
+    /// loop. Loop sleeps for the duration of the channel heartbeat, then
+    /// sends a ping message with a random nonce. Loop starts a timer, waits
+    /// for the pong reply and insures the nonce is the same.
     async fn run_ping_pong(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net", "ProtocolPing::run_ping_pong() [START]");
         // Creates a subscription to pong message.
@@ -85,7 +85,8 @@ impl ProtocolPing {
         }
     }
 
-    /// Waits for ping, then replies with pong. Copies ping's nonce into the pong reply.
+    /// Waits for ping, then replies with pong. Copies ping's nonce into the
+    /// pong reply.
     async fn reply_to_ping(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net", "ProtocolPing::reply_to_ping() [START]");
         // Creates a subscription to ping message.

src/net/protocols/protocol_seed.rs

@@ -24,9 +24,9 @@ impl ProtocolSeed {
         })
     }
 
-    /// Starts the seed protocol. Creates a subscription to the address message, then
-    /// sends our address to the seed server. Sends a get-address message and receives
-    /// an address message.
+    /// Starts the seed protocol. Creates a subscription to the address message,
+    /// then sends our address to the seed server. Sends a get-address
+    /// message and receives an address message.
     pub async fn start(self: Arc<Self>, _executor: Arc<Executor<'_>>) -> NetResult<()> {
         debug!(target: "net", "ProtocolSeed::start() [START]");
         // Create a subscription to address message.
@@ -53,8 +53,9 @@ impl ProtocolSeed {
         Ok(())
     }
 
-    /// Sends own external address over a channel. Imports own external address from
-    /// settings, then adds that address to an address message and sends it out over the channel.
+    /// Sends own external address over a channel. Imports own external address
+    /// from settings, then adds that address to an address message and
+    /// sends it out over the channel.
     pub async fn send_self_address(&self) -> NetResult<()> {
         match self.settings.external_addr {
             Some(addr) => {

src/net/protocols/protocol_version.rs

@@ -9,7 +9,8 @@ use crate::net::messages;
 use crate::net::utility::sleep;
 use crate::net::{ChannelPtr, SettingsPtr};
 
-/// Protocol for version information handshake between nodes at the start of a connection.
+/// Protocol for version information handshake between nodes at the start of a
+/// connection.
 pub struct ProtocolVersion {
     channel: ChannelPtr,
     version_sub: MessageSubscription<messages::VersionMessage>,
@@ -18,8 +19,9 @@ pub struct ProtocolVersion {
 }
 
 impl ProtocolVersion {
-    /// Create a new version protocol. Makes a version and version acknowledgement
-    /// subscription, then adds them to a version protocol instance.
+    /// Create a new version protocol. Makes a version and version
+    /// acknowledgement subscription, then adds them to a version protocol
+    /// instance.
     pub async fn new(channel: ChannelPtr, settings: SettingsPtr) -> Arc<Self> {
         // Creates a version subscription.
         let version_sub = channel
@@ -42,8 +44,9 @@ impl ProtocolVersion {
             settings,
         })
     }
-    /// Start version information exchange. Start the timer. Send version info and
-    /// wait for version acknowledgement. Wait for version info and send version acknowledgement.
+    /// Start version information exchange. Start the timer. Send version info
+    /// and wait for version acknowledgement. Wait for version info and send
+    /// version acknowledgement.
     pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         debug!(target: "net", "ProtocolVersion::run() [START]");
         // Start timer
@@ -80,7 +83,8 @@ impl ProtocolVersion {
         debug!(target: "net", "ProtocolVersion::send_version() [END]");
         Ok(())
     }
-    /// Recieve version info, check the message is okay and send version acknowledgement.
+    /// Recieve version info, check the message is okay and send version
+    /// acknowledgement.
     async fn recv_version(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net", "ProtocolVersion::recv_version() [START]");
         // Rec

src/net/sessions/inbound_session.rs

@@ -28,8 +28,9 @@ impl InboundSession {
             accept_task: StoppableTask::new(),
         })
     }
-    /// Starts the inbound session. Begins by accepting connections and fails if the
-    /// address is not configured. Then runs the channel subscription loop.
+    /// Starts the inbound session. Begins by accepting connections and fails if
+    /// the address is not configured. Then runs the channel subscription
+    /// loop.
     pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         match self.p2p().settings().inbound {
             Some(accept_addr) => {
@@ -71,7 +72,8 @@ impl InboundSession {
         result
     }
 
-    /// Wait for all new channels created by the acceptor and call setup_channel() on them.
+    /// Wait for all new channels created by the acceptor and call
+    /// setup_channel() on them.
     async fn channel_sub_loop(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         let channel_sub = self.acceptor.clone().subscribe().await;
         loop {
@@ -85,7 +87,8 @@ impl InboundSession {
     }
 
     /// Registers the channel. First performs a network handshake and starts the
-    /// channel. Then starts sending keep-alive and address messages across the channel.
+    /// channel. Then starts sending keep-alive and address messages across the
+    /// channel.
     async fn setup_channel(
         self: Arc<Self>,
         channel: ChannelPtr,

src/net/sessions/outbound_session.rs

@@ -56,10 +56,11 @@ impl OutboundSession {
         }
     }
 
-    /// Start making outbound connections. Creates a connector object, then starts a
-    /// connect loop. Loads a valid address then tries to connect. Once connected,
-    /// registers the channel, removes it from the list of pending channels, and
-    /// starts sending messages across the channel. Otherwise returns a network error.
+    /// Start making outbound connections. Creates a connector object, then
+    /// starts a connect loop. Loads a valid address then tries to connect.
+    /// Once connected, registers the channel, removes it from the list of
+    /// pending channels, and starts sending messages across the channel.
+    /// Otherwise returns a network error.
     pub async fn channel_connect_loop(
         self: Arc<Self>,
         slot_number: u32,
@@ -102,10 +103,11 @@ impl OutboundSession {
         }
     }
 
-    /// Loops through host addresses to find a outbound address that we can connect
-    /// to. Checks whether address is valid by making sure it isn't our own inbound
-    /// address, then checks whether it is already connected (exists) or connecting
-    /// (pending). Keeps looping until address is found that passes all checks.
+    /// Loops through host addresses to find a outbound address that we can
+    /// connect to. Checks whether address is valid by making sure it isn't
+    /// our own inbound address, then checks whether it is already connected
+    /// (exists) or connecting (pending). Keeps looping until address is
+    /// found that passes all checks.
     async fn load_address(&self, slot_number: u32) -> NetResult<SocketAddr> {
         let p2p = self.p2p();
         let hosts = p2p.hosts();
@@ -140,7 +142,8 @@ impl OutboundSession {
         }
     }
 
-    /// Checks whether an address is our own inbound address to avoid connecting to ourselves.
+    /// Checks whether an address is our own inbound address to avoid connecting
+    /// to ourselves.
     fn is_self_inbound(addr: &SocketAddr, inbound_addr: &Option<SocketAddr>) -> bool {
         match inbound_addr {
             Some(inbound_addr) => inbound_addr == addr,

src/net/sessions/seed_session.rs

@@ -70,8 +70,9 @@ impl SeedSession {
         Ok(())
     }
 
-    /// Connects to a seed socket address. Registers a new channel with a network
-    /// handshake, then starts the keep-alive messages and seed protocol.
+    /// Connects to a seed socket address. Registers a new channel with a
+    /// network handshake, then starts the keep-alive messages and seed
+    /// protocol.
     async fn start_seed(
         self: Arc<Self>,
         seed_index: usize,

src/net/sessions/session.rs

@@ -8,7 +8,8 @@ use crate::net::p2p::P2pPtr;
 use crate::net::protocols::ProtocolVersion;
 use crate::net::ChannelPtr;
 
-/// Removes channel from the list of connected channels when a stop signal is received.
+/// Removes channel from the list of connected channels when a stop signal is
+/// received.
 async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
     debug!(target: "net", "remove_sub_on_stop() [START]");
     // Subscribe to stop events
@@ -29,8 +30,8 @@ async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
 /// registering the channel and initializing the channel by performing a network
 /// handshake.
 pub trait Session: Sync {
-    /// Registers a new channel with the session. Performs a network handshake and
-    /// starts the channel.
+    /// Registers a new channel with the session. Performs a network handshake
+    /// and starts the channel.
     async fn register_channel(
         self: Arc<Self>,
         channel: ChannelPtr,
@@ -51,9 +52,9 @@ pub trait Session: Sync {
         Ok(())
     }
 
-    /// Performs network handshake to initialize channel. Adds the channel to the
-    /// list of connected channels, and prepares to remove the channel when a stop
-    /// signal is received.
+    /// Performs network handshake to initialize channel. Adds the channel to
+    /// the list of connected channels, and prepares to remove the channel
+    /// when a stop signal is received.
     async fn perform_handshake_protocols(
         &self,
         protocol_version: Arc<ProtocolVersion>,