tidy up the code

src/bin/dfi.rs

@@ -93,7 +93,7 @@ impl RpcInterface {
     }
 
     async fn serve(self: Arc<Self>, mut req: Request) -> http_types::Result<Response> {
-        println!("Serving {}", req.url());
+        info!("RPC serving {}", req.url());
 
         let request = req.body_string().await?;
 
@@ -138,9 +138,6 @@ impl RpcInterface {
 }
 
 async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<()> {
-    //sapvi::net::message_subscriber::doteste().await;
-    //return Ok(());
-
     let p2p = net::P2p::new(options.network_settings);
 
     let rpc = RpcInterface::new(p2p.clone());

src/error.rs

@@ -142,7 +142,7 @@ impl From<NetError> for Error {
 }
 
 impl From<std::string::FromUtf8Error> for Error {
-    fn from(err: std::string::FromUtf8Error) -> Error {
+    fn from(_err: std::string::FromUtf8Error) -> Error {
         Error::Utf8Error
     }
 }

src/lib.rs

@@ -9,7 +9,6 @@ pub mod error;
 pub mod net;
 pub mod serial;
 pub mod system;
-pub mod utility;
 pub mod vm;
 pub mod vm_serial;
 

src/net/channel.rs

@@ -11,9 +11,7 @@ use std::sync::Arc;
 
 use crate::error;
 use crate::net::error::{NetError, NetResult};
-use crate::net::message_subscriber::{
-    MessageSubsystem, MessageSubscription, Message
-};
+use crate::net::message_subscriber::{MessageSubscription, MessageSubsystem};
 use crate::net::messages;
 use crate::net::settings::SettingsPtr;
 use crate::system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription};
@@ -32,7 +30,11 @@ pub struct Channel {
 }
 
 impl Channel {
-    pub async fn new(stream: Async<TcpStream>, address: SocketAddr, settings: SettingsPtr) -> Arc<Self> {
+    pub async fn new(
+        stream: Async<TcpStream>,
+        address: SocketAddr,
+        settings: SettingsPtr,
+    ) -> Arc<Self> {
         let (reader, writer) = stream.split();
         let reader = Mutex::new(reader);
         let writer = Mutex::new(writer);
@@ -90,7 +92,7 @@ impl Channel {
         sub
     }
 
-    pub async fn send<M: Message>(&self, message: M) -> NetResult<()> {
+    pub async fn send<M: messages::Message>(&self, message: M) -> NetResult<()> {
         debug!(target: "net",
             "Channel::send() [START, command={:?}, address={}]",
             M::name(),
@@ -117,21 +119,19 @@ impl Channel {
         result
     }
 
-    async fn send_message<M: Message>(&self, message: M) -> error::Result<()> {
+    async fn send_message<M: messages::Message>(&self, message: M) -> error::Result<()> {
         let mut payload = Vec::new();
         message.encode(&mut payload)?;
         let packet = messages::Packet {
             command: String::from(M::name()),
             payload,
         };
-        
+
         let stream = &mut *self.writer.lock().await;
         messages::send_packet(stream, packet).await
     }
 
-    pub async fn subscribe_msg<M: Message>(
-        &self
-    ) -> NetResult<MessageSubscription<M>> {
+    pub async fn subscribe_msg<M: messages::Message>(&self) -> NetResult<MessageSubscription<M>> {
         debug!(target: "net",
             "Channel::subscribe_msg() [START, command={:?}, address={}]",
             M::name(),
@@ -158,12 +158,24 @@ impl Channel {
     }
 
     async fn setup_dispatchers(message_subsystem: &MessageSubsystem) {
-        message_subsystem.add_dispatch::<messages::VersionMessage>().await;
-        message_subsystem.add_dispatch::<messages::VerackMessage>().await;
-        message_subsystem.add_dispatch::<messages::PingMessage>().await;
-        message_subsystem.add_dispatch::<messages::PongMessage>().await;
-        message_subsystem.add_dispatch::<messages::GetAddrsMessage>().await;
-        message_subsystem.add_dispatch::<messages::AddrsMessage>().await;
+        message_subsystem
+            .add_dispatch::<messages::VersionMessage>()
+            .await;
+        message_subsystem
+            .add_dispatch::<messages::VerackMessage>()
+            .await;
+        message_subsystem
+            .add_dispatch::<messages::PingMessage>()
+            .await;
+        message_subsystem
+            .add_dispatch::<messages::PongMessage>()
+            .await;
+        message_subsystem
+            .add_dispatch::<messages::GetAddrsMessage>()
+            .await;
+        message_subsystem
+            .add_dispatch::<messages::AddrsMessage>()
+            .await;
     }
 
     async fn main_receive_loop(self: Arc<Self>) -> NetResult<()> {
@@ -193,7 +205,9 @@ impl Channel {
             };
 
             // Send result to our subscribers
-            self.message_subsystem.notify(&packet.command, packet.payload).await;
+            self.message_subsystem
+                .notify(&packet.command, packet.payload)
+                .await;
         }
     }
 

src/net/message_subscriber.rs

@@ -10,16 +10,12 @@ use std::sync::Arc;
 
 use crate::error::Result;
 use crate::net::error::{NetError, NetResult};
+use crate::net::messages::Message;
 use crate::serial::{Decodable, Encodable};
 
-
 pub type MessageSubscriptionID = u64;
 type MessageResult<M> = NetResult<Arc<M>>;
 
-pub trait Message: 'static + Encodable + Decodable + Send + Sync {
-    fn name() -> &'static str;
-}
-
 pub struct MessageSubscription<M: Message> {
     id: MessageSubscriptionID,
     recv_queue: async_channel::Receiver<MessageResult<M>>,
@@ -131,7 +127,7 @@ impl<M: Message> MessageDispatcherInterface for MessageDispatcher<M> {
             Ok(message) => {
                 let message = Ok(Arc::new(message));
                 self.trigger_all(message).await
-            },
+            }
             Err(err) => {
                 error!("Unable to decode data. Dropping...: {}", err);
             }
@@ -147,70 +143,6 @@ impl<M: Message> MessageDispatcherInterface for MessageDispatcher<M> {
     }
 }
 
-use crate::net::messages::{PingMessage, PongMessage, GetAddrsMessage, AddrsMessage, VersionMessage, VerackMessage};
-
-impl Message for PingMessage {
-    fn name() -> &'static str {
-        "ping"
-    }
-}
-
-impl Message for PongMessage {
-    fn name() -> &'static str {
-        "pong"
-    }
-}
-
-impl Message for GetAddrsMessage {
-    fn name() -> &'static str {
-        "getaddr"
-    }
-}
-
-impl Message for AddrsMessage {
-    fn name() -> &'static str {
-        "addr"
-    }
-}
-
-impl Message for VersionMessage {
-    fn name() -> &'static str {
-        "version"
-    }
-}
-
-impl Message for VerackMessage {
-    fn name() -> &'static str {
-        "verack"
-    }
-}
-
-struct MyVersionMessage {
-    x: u32,
-}
-
-impl Message for MyVersionMessage {
-    fn name() -> &'static str {
-        "verver"
-    }
-}
-
-impl Encodable for MyVersionMessage {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-        let mut len = 0;
-        len += self.x.encode(&mut s)?;
-        Ok(len)
-    }
-}
-
-impl Decodable for MyVersionMessage {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-        Ok(Self {
-            x: Decodable::decode(&mut d)?,
-        })
-    }
-}
-
 pub struct MessageSubsystem {
     dispatchers: Mutex<HashMap<&'static str, Arc<dyn MessageDispatcherInterface>>>,
 }
@@ -230,12 +162,7 @@ impl MessageSubsystem {
     }
 
     pub async fn subscribe<M: Message>(&self) -> NetResult<MessageSubscription<M>> {
-        let dispatcher = self
-            .dispatchers
-            .lock()
-            .await
-            .get(M::name())
-            .cloned();
+        let dispatcher = self.dispatchers.lock().await.get(M::name()).cloned();
 
         let sub = match dispatcher {
             Some(dispatcher) => {
@@ -281,6 +208,31 @@ impl MessageSubsystem {
 }
 
 pub async fn doteste() {
+    struct MyVersionMessage {
+        x: u32,
+    }
+
+    impl Message for MyVersionMessage {
+        fn name() -> &'static str {
+            "verver"
+        }
+    }
+
+    impl Encodable for MyVersionMessage {
+        fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+            let mut len = 0;
+            len += self.x.encode(&mut s)?;
+            Ok(len)
+        }
+    }
+
+    impl Decodable for MyVersionMessage {
+        fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+            Ok(Self {
+                x: Decodable::decode(&mut d)?,
+            })
+        }
+    }
     println!("hello");
 
     let subsystem = MessageSubsystem::new();

src/net/messages.rs

@@ -1,22 +1,18 @@
 use futures::prelude::*;
 use log::*;
-use num_enum::{IntoPrimitive, TryFromPrimitive};
-use smol::Executor;
-use smol::Timer;
-use std::convert::TryFrom;
 use std::io;
-use std::io::Cursor;
 use std::net::SocketAddr;
-use std::sync::Arc;
-use std::time::Duration;
 
-use crate::async_serial::{AsyncReadExt, AsyncWriteExt};
 use crate::error::{Error, Result};
 pub use crate::net::AsyncTcpStream;
-use crate::serial::{serialize, Decodable, Encodable, VarInt};
+use crate::serial::{Decodable, Encodable, VarInt};
 
 const MAGIC_BYTES: [u8; 4] = [0xd9, 0xef, 0xb6, 0x7d];
 
+pub trait Message: 'static + Encodable + Decodable + Send + Sync {
+    fn name() -> &'static str;
+}
+
 pub struct PingMessage {
     pub nonce: u32,
 }
@@ -35,6 +31,42 @@ pub struct VersionMessage {}
 
 pub struct VerackMessage {}
 
+impl Message for PingMessage {
+    fn name() -> &'static str {
+        "ping"
+    }
+}
+
+impl Message for PongMessage {
+    fn name() -> &'static str {
+        "pong"
+    }
+}
+
+impl Message for GetAddrsMessage {
+    fn name() -> &'static str {
+        "getaddr"
+    }
+}
+
+impl Message for AddrsMessage {
+    fn name() -> &'static str {
+        "addr"
+    }
+}
+
+impl Message for VersionMessage {
+    fn name() -> &'static str {
+        "version"
+    }
+}
+
+impl Message for VerackMessage {
+    fn name() -> &'static str {
+        "verack"
+    }
+}
+
 impl Encodable for PingMessage {
     fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
         let mut len = 0;
@@ -156,7 +188,10 @@ pub async fn read_packet<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Packet>
     }
     debug!(target: "net", "read payload {} bytes", payload_len);
 
-    Ok(Packet { command: command, payload })
+    Ok(Packet {
+        command: command,
+        payload,
+    })
 }
 
 pub async fn send_packet<W: AsyncWrite + Unpin>(stream: &mut W, packet: Packet) -> Result<()> {
@@ -183,4 +218,3 @@ pub async fn send_packet<W: AsyncWrite + Unpin>(stream: &mut W, packet: Packet)
 
     Ok(())
 }
-

src/net/protocols/protocol_ping.rs

@@ -2,7 +2,7 @@ use log::*;
 use rand::Rng;
 use smol::Executor;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 
 use crate::net::error::{NetError, NetResult};
 use crate::net::messages;

src/utility.rs

@@ -1,54 +0,0 @@
-use std::collections::HashMap;
-use std::fs::OpenOptions;
-use std::io::prelude::*;
-use std::net::SocketAddr;
-use std::sync::atomic::AtomicU64;
-use std::sync::Arc;
-use std::time::{SystemTime, UNIX_EPOCH};
-
-use rand::seq::SliceRandom;
-use smol::{Executor, Task};
-
-//use crate::{net, serial, Channel, ClientProtocol, Result, SlabsManagerSafe};
-use crate::{net::messages as net, serial, Result};
-
-pub type AddrsStorage = std::sync::Arc<async_std::sync::Mutex<Vec<SocketAddr>>>;
-
-pub type Clock = std::sync::Arc<AtomicU64>;
-
-pub fn get_current_time() -> u64 {
-    let start = SystemTime::now();
-    let since_the_epoch = start
-        .duration_since(UNIX_EPOCH)
-        .expect("Incorrect system clock: time went backwards");
-    let in_ms =
-        since_the_epoch.as_secs() * 1000 + since_the_epoch.subsec_nanos() as u64 / 1_000_000;
-    return in_ms;
-}
-
-pub fn save_to_addrs_store(stored_addrs: &Vec<SocketAddr>) -> Result<()> {
-    let mut writer = OpenOptions::new()
-        .write(true)
-        .create(true)
-        .open("addrs.dps")?;
-    let buffer = serial::serialize(stored_addrs);
-    writer.write_all(&buffer)?;
-    Ok(())
-}
-
-pub fn load_stored_addrs() -> Result<Vec<SocketAddr>> {
-    let mut reader = OpenOptions::new()
-        .read(true)
-        .write(true)
-        .create(true)
-        .open("addrs.dps")?;
-    let mut buffer = Vec::new();
-    reader.read_to_end(&mut buffer)?;
-    if !buffer.is_empty() {
-        let addrs: Vec<SocketAddr> = serial::deserialize(&buffer)?;
-        Ok(addrs)
-    } else {
-        Ok(vec![])
-    }
-}
-