cargo fmt and fix

src/async_serial.rs

@@ -2,7 +2,6 @@ use futures::prelude::*;
 
 use crate::endian;
 use crate::error::{Error, Result};
-use crate::net::AsyncTcpStream;
 use crate::serial::VarInt;
 
 impl VarInt {

src/bin/dfi.rs

@@ -3,20 +3,16 @@ extern crate clap;
 use async_executor::Executor;
 use async_std::sync::Mutex;
 use easy_parallel::Parallel;
-use log::*;
 use serde_json::json;
-use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
-
-use sapvi::{net, Result};
-
 use std::net::TcpListener;
-
 use async_native_tls::TlsAcceptor;
 use http_types::{Request, Response, StatusCode};
 use smol::Async;
 
+use sapvi::{net, Result};
+
 /// Listens for incoming connections and serves them.
 async fn listen(
     executor: Arc<Executor<'_>>,

src/error.rs

@@ -138,4 +138,3 @@ impl From<NetError> for Error {
         }
     }
 }
-

src/net/acceptor.rs

@@ -1,4 +1,3 @@
-use futures::FutureExt;
 use log::*;
 use smol::{Async, Executor};
 use std::net::{SocketAddr, TcpListener};
@@ -47,8 +46,7 @@ impl Acceptor {
         self.channel_subscriber.clone().subscribe().await
     }
 
-    fn setup(
-        accept_addr: SocketAddr) -> NetResult<Async<TcpListener>> {
+    fn setup(accept_addr: SocketAddr) -> NetResult<Async<TcpListener>> {
         let listener = match Async::<TcpListener>::bind(accept_addr) {
             Ok(l) => l,
             Err(err) => {
@@ -73,7 +71,7 @@ impl Acceptor {
             self.clone().run_accept_loop(listener),
             |result| self.handle_stop(result),
             NetError::ServiceStopped,
-            executor
+            executor,
         );
     }
 

src/net/channel.rs

@@ -4,9 +4,9 @@ use futures::AsyncReadExt;
 use futures::FutureExt;
 use log::*;
 use smol::{Async, Executor};
-use std::future::Future;
+
 use std::net::{SocketAddr, TcpStream};
-use std::pin::Pin;
+
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 
@@ -16,7 +16,6 @@ use crate::net::message_subscriber::{
 };
 use crate::net::messages;
 use crate::net::settings::SettingsPtr;
-use crate::net::utility::clone_net_error;
 use crate::system::{Subscriber, SubscriberPtr, Subscription};
 
 pub type ChannelPtr = Arc<Channel>;

src/net/connector.rs

@@ -1,8 +1,6 @@
 use futures::FutureExt;
-use log::*;
-use smol::{Async, Executor};
+use smol::{Async};
 use std::net::{SocketAddr, TcpStream};
-use std::sync::Arc;
 
 use crate::net::error::{NetError, NetResult};
 use crate::net::utility::sleep;

src/net/error.rs

@@ -26,4 +26,3 @@ impl fmt::Display for NetError {
         }
     }
 }
-

src/net/message_subscriber.rs

@@ -5,7 +5,6 @@ use std::sync::Arc;
 
 use crate::net::error::NetResult;
 use crate::net::messages::{Message, PacketType};
-use crate::net::utility::clone_net_error;
 
 pub type MessageSubscriberPtr = Arc<MessageSubscriber>;
 

src/net/messages.rs

@@ -2,11 +2,11 @@ use futures::prelude::*;
 use log::*;
 use num_enum::{IntoPrimitive, TryFromPrimitive};
 use smol::Executor;
-use smol::{Async, Timer};
+use smol::Timer;
 use std::convert::TryFrom;
 use std::io;
 use std::io::Cursor;
-use std::net::{SocketAddr, TcpStream};
+use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -49,11 +49,11 @@ pub enum Message {
 }
 
 pub struct PingMessage {
-    pub nonce: u32
+    pub nonce: u32,
 }
 
 pub struct PongMessage {
-    pub nonce: u32
+    pub nonce: u32,
 }
 
 pub struct GetAddrsMessage {}
@@ -192,25 +192,25 @@ impl Decodable for AddrsMessage {
 }
 
 impl Encodable for VersionMessage {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+    fn encode<S: io::Write>(&self, _s: S) -> Result<usize> {
         Ok(0)
     }
 }
 
 impl Decodable for VersionMessage {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+    fn decode<D: io::Read>(_d: D) -> Result<Self> {
         Ok(Self {})
     }
 }
 
 impl Encodable for VerackMessage {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+    fn encode<S: io::Write>(&self, _s: S) -> Result<usize> {
         Ok(0)
     }
 }
 
 impl Decodable for VerackMessage {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+    fn decode<D: io::Read>(_d: D) -> Result<Self> {
         Ok(Self {})
     }
 }
@@ -218,15 +218,15 @@ impl Decodable for VerackMessage {
 impl Message {
     pub fn packet_type(&self) -> PacketType {
         match self {
-            Message::Ping(message) => PacketType::Ping,
-            Message::Pong(message) => PacketType::Pong,
-            Message::GetAddrs(message) => PacketType::GetAddrs,
-            Message::Addrs(message) => PacketType::Addrs,
-            Message::Inv(message) => PacketType::Inv,
-            Message::GetSlabs(message) => PacketType::GetSlabs,
-            Message::Slab(message) => PacketType::Slab,
-            Message::Version(message) => PacketType::Version,
-            Message::Verack(message) => PacketType::Verack,
+            Message::Ping(_message) => PacketType::Ping,
+            Message::Pong(_message) => PacketType::Pong,
+            Message::GetAddrs(_message) => PacketType::GetAddrs,
+            Message::Addrs(_message) => PacketType::Addrs,
+            Message::Inv(_message) => PacketType::Inv,
+            Message::GetSlabs(_message) => PacketType::GetSlabs,
+            Message::Slab(_message) => PacketType::Slab,
+            Message::Version(_message) => PacketType::Version,
+            Message::Verack(_message) => PacketType::Verack,
         }
     }
 

src/net/protocols/mod.rs

@@ -1,11 +1,11 @@
 pub mod protocol_address;
+pub mod protocol_jobs_manager;
 pub mod protocol_ping;
 pub mod protocol_seed;
 pub mod protocol_version;
-pub mod protocol_jobs_manager;
 
 pub use protocol_address::ProtocolAddress;
+pub use protocol_jobs_manager::{ProtocolJobsManager, ProtocolJobsManagerPtr};
 pub use protocol_ping::ProtocolPing;
 pub use protocol_seed::ProtocolSeed;
 pub use protocol_version::ProtocolVersion;
-pub use protocol_jobs_manager::{ProtocolJobsManager, ProtocolJobsManagerPtr};

src/net/protocols/protocol_address.rs

@@ -1,28 +1,24 @@
-use futures::FutureExt;
-use log::*;
-use rand::Rng;
-use smol::{Executor, Task};
+use smol::Executor;
 use std::sync::Arc;
 
-use crate::net::error::{NetError, NetResult};
-use crate::net::messages;
-use crate::net::utility::sleep;
-use crate::net::{ChannelPtr, SettingsPtr};
 use crate::net::protocols::{ProtocolJobsManager, ProtocolJobsManagerPtr};
+use crate::net::{ChannelPtr, SettingsPtr};
 
 pub struct ProtocolAddress {
     channel: ChannelPtr,
     settings: SettingsPtr,
 
-    jobsman: ProtocolJobsManagerPtr
+    jobsman: ProtocolJobsManagerPtr,
 }
 
 impl ProtocolAddress {
     pub fn new(channel: ChannelPtr, settings: SettingsPtr) -> Arc<Self> {
-        Arc::new(Self { channel: channel.clone(), settings, jobsman: ProtocolJobsManager::new(channel) })
+        Arc::new(Self {
+            channel: channel.clone(),
+            settings,
+            jobsman: ProtocolJobsManager::new(channel),
+        })
     }
 
-    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
-    }
+    pub async fn start(self: Arc<Self>, _executor: Arc<Executor<'_>>) {}
 }
-

src/net/protocols/protocol_jobs_manager.rs

@@ -1,7 +1,7 @@
-use std::sync::Arc;
-use smol::Task;
-use futures::Future;
 use async_std::sync::Mutex;
+use futures::Future;
+use smol::Task;
+use std::sync::Arc;
 
 use crate::net::error::NetResult;
 use crate::net::ChannelPtr;
@@ -11,14 +11,14 @@ pub type ProtocolJobsManagerPtr = Arc<ProtocolJobsManager>;
 
 pub struct ProtocolJobsManager {
     channel: ChannelPtr,
-    tasks: Mutex<Vec<Task<NetResult<()>>>>
+    tasks: Mutex<Vec<Task<NetResult<()>>>>,
 }
 
 impl ProtocolJobsManager {
     pub fn new(channel: ChannelPtr) -> Arc<Self> {
         Arc::new(Self {
             channel,
-            tasks: Mutex::new(Vec::new())
+            tasks: Mutex::new(Vec::new()),
         })
     }
 
@@ -28,7 +28,7 @@ impl ProtocolJobsManager {
 
     pub async fn spawn<'a, F>(&self, future: F, executor: ExecutorPtr<'a>)
     where
-        F: Future<Output=NetResult<()>> + Send + 'a
+        F: Future<Output = NetResult<()>> + Send + 'a,
     {
         self.tasks.lock().await.push(executor.spawn(future))
     }
@@ -50,4 +50,3 @@ impl ProtocolJobsManager {
         }
     }
 }
-

src/net/protocols/protocol_ping.rs

@@ -1,31 +1,40 @@
-use futures::FutureExt;
 use log::*;
 use rand::Rng;
-use smol::{Executor, Task};
+use smol::{Executor};
 use std::sync::Arc;
 
 use crate::net::error::{NetError, NetResult};
 use crate::net::messages;
+use crate::net::protocols::{ProtocolJobsManager, ProtocolJobsManagerPtr};
 use crate::net::utility::sleep;
 use crate::net::{ChannelPtr, SettingsPtr};
-use crate::net::protocols::{ProtocolJobsManager, ProtocolJobsManagerPtr};
 
 pub struct ProtocolPing {
     channel: ChannelPtr,
     settings: SettingsPtr,
 
-    jobsman: ProtocolJobsManagerPtr
+    jobsman: ProtocolJobsManagerPtr,
 }
 
 impl ProtocolPing {
     pub fn new(channel: ChannelPtr, settings: SettingsPtr) -> Arc<Self> {
-        Arc::new(Self { channel: channel.clone(), settings, jobsman: ProtocolJobsManager::new(channel) })
+        Arc::new(Self {
+            channel: channel.clone(),
+            settings,
+            jobsman: ProtocolJobsManager::new(channel),
+        })
     }
 
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
         self.jobsman.clone().start(executor.clone());
-        self.jobsman.clone().spawn(self.clone().run_ping_pong(), executor.clone()).await;
-        self.jobsman.clone().spawn(self.reply_to_ping(), executor).await;
+        self.jobsman
+            .clone()
+            .spawn(self.clone().run_ping_pong(), executor.clone())
+            .await;
+        self.jobsman
+            .clone()
+            .spawn(self.reply_to_ping(), executor)
+            .await;
     }
 
     async fn run_ping_pong(self: Arc<Self>) -> NetResult<()> {
@@ -43,9 +52,7 @@ impl ProtocolPing {
             let nonce = Self::random_nonce();
 
             // Send ping message
-            let ping = messages::Message::Ping(messages::PingMessage {
-                nonce
-            });
+            let ping = messages::Message::Ping(messages::PingMessage { nonce });
             self.channel.clone().send(ping).await?;
 
             // Wait for pong, check nonce matches
@@ -70,9 +77,7 @@ impl ProtocolPing {
             let ping = receive_message!(ping_sub, messages::Message::Ping);
 
             // Send ping message
-            let pong = messages::Message::Pong(messages::PongMessage {
-                nonce: ping.nonce
-            });
+            let pong = messages::Message::Pong(messages::PongMessage { nonce: ping.nonce });
             self.channel.clone().send(pong).await?;
         }
     }

src/net/protocols/protocol_seed.rs

@@ -1,9 +1,7 @@
-use futures::FutureExt;
-use owning_ref::OwningRef;
 use smol::Executor;
 use std::sync::Arc;
 
-use crate::net::error::{NetError, NetResult};
+use crate::net::error::NetResult;
 use crate::net::messages;
 use crate::net::{ChannelPtr, HostsPtr, SettingsPtr};
 
@@ -22,7 +20,7 @@ impl ProtocolSeed {
         })
     }
 
-    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
+    pub async fn start(self: Arc<Self>, _executor: Arc<Executor<'_>>) -> NetResult<()> {
         let addr_sub = self
             .channel
             .clone()

src/net/protocols/protocol_version.rs

@@ -4,7 +4,7 @@ use std::sync::Arc;
 
 use crate::net::error::{NetError, NetResult};
 use crate::net::messages;
-use crate::net::utility::{clone_net_error, sleep};
+use crate::net::utility::sleep;
 use crate::net::{ChannelPtr, SettingsPtr};
 
 pub struct ProtocolVersion {
@@ -50,7 +50,7 @@ impl ProtocolVersion {
             .subscribe_msg(messages::PacketType::Version)
             .await;
 
-        let version_msg = version_sub.receive().await?;
+        let _version_msg = version_sub.receive().await?;
 
         // Check the message is OK
 

src/net/proxy.rs

@@ -1,5 +1,5 @@
-use smol::{Async, Executor};
-use std::net::{SocketAddr, TcpStream};
+use smol::{Async};
+use std::net::{TcpStream};
 
 pub struct Proxy {
     stream: Async<TcpStream>,

src/net/sessions/inbound_session.rs

@@ -4,10 +4,10 @@ use std::net::SocketAddr;
 use std::sync::{Arc, Weak};
 
 use crate::net::error::{NetError, NetResult};
-use crate::net::protocols::{ProtocolPing, ProtocolAddress, ProtocolSeed};
+use crate::net::protocols::{ProtocolAddress, ProtocolPing};
 use crate::net::sessions::Session;
 use crate::net::{Acceptor, AcceptorPtr};
-use crate::net::{ChannelPtr, Connector, HostsPtr, P2p, SettingsPtr};
+use crate::net::{ChannelPtr, P2p, SettingsPtr};
 use crate::system::{StoppableTask, StoppableTaskPtr};
 
 pub struct InboundSession {
@@ -25,13 +25,18 @@ impl InboundSession {
 
         let acceptor = Acceptor::new(settings);
 
-        Arc::new(Self { p2p, acceptor, accept_task: StoppableTask::new() })
+        Arc::new(Self {
+            p2p,
+            acceptor,
+            accept_task: StoppableTask::new(),
+        })
     }
 
     pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         match self.p2p().settings().inbound {
             Some(accept_addr) => {
-                self.clone().start_accept_session(accept_addr, executor.clone())?;
+                self.clone()
+                    .start_accept_session(accept_addr, executor.clone())?;
             }
             None => {
                 info!("Not configured for accepting incoming connections.");
@@ -42,9 +47,10 @@ impl InboundSession {
         self.accept_task.clone().start(
             self.clone().channel_sub_loop(executor.clone()),
             // Ignore stop handler
-            |_| { async {} },
+            |_| async {},
             NetError::ServiceStopped,
-            executor);
+            executor,
+        );
 
         Ok(())
     }
@@ -60,7 +66,7 @@ impl InboundSession {
     ) -> NetResult<()> {
         info!("Starting inbound session on {}", accept_addr);
         let result = self.acceptor.clone().start(accept_addr, executor);
-        if let Err(err) = result  {
+        if let Err(err) = result {
             error!("Error starting listener: {}", err);
         }
         result
@@ -72,11 +78,17 @@ impl InboundSession {
             let channel = (*channel_sub.receive().await).clone()?;
             // Spawn a detached task to process the channel
             // This will just perform the channel setup then exit.
-            executor.spawn(self.clone().setup_channel(channel, executor.clone())).detach();
+            executor
+                .spawn(self.clone().setup_channel(channel, executor.clone()))
+                .detach();
         }
     }
 
-    async fn setup_channel(self: Arc<Self>, channel: ChannelPtr, executor: Arc<Executor<'_>>) -> NetResult<()> {
+    async fn setup_channel(
+        self: Arc<Self>,
+        channel: ChannelPtr,
+        executor: Arc<Executor<'_>>,
+    ) -> NetResult<()> {
         info!("Connected inbound [{}]", channel.address());
 
         self.clone()
@@ -85,8 +97,7 @@ impl InboundSession {
 
         let settings = self.p2p.upgrade().unwrap().settings();
 
-        self.attach_protocols(channel, settings, executor)
-            .await
+        self.attach_protocols(channel, settings, executor).await
     }
 
     async fn attach_protocols(

src/net/utility.rs

@@ -4,4 +4,3 @@ use std::time::Duration;
 pub async fn sleep(seconds: u32) {
     Timer::after(Duration::from_secs(seconds.into())).await;
 }
-

src/system/mod.rs

@@ -5,4 +5,3 @@ pub mod types;
 pub use stoppable_task::{StoppableTask, StoppableTaskPtr};
 pub use subscriber::{Subscriber, SubscriberPtr, Subscription};
 pub use types::ExecutorPtr;
-

src/system/stoppable_task.rs

@@ -34,7 +34,7 @@ impl StoppableTask {
         MainFut: Future<Output = std::result::Result<(), Error>> + Send + 'a,
         StopFut: Future<Output = ()> + Send,
         StopFn: FnOnce(std::result::Result<(), Error>) -> StopFut + Send + 'a,
-        Error: std::error::Error + Send + 'a
+        Error: std::error::Error + Send + 'a,
     {
         executor
             .spawn(async move {

src/system/types.rs

@@ -2,4 +2,3 @@ use smol::Executor;
 use std::sync::Arc;
 
 pub type ExecutorPtr<'a> = Arc<Executor<'a>>;
-