Merge remote-tracking branch 'origin/master' into feature/lisp

.github/workflows/matrix-commit-message.yml

@@ -0,0 +1,40 @@
+on: [push]
+
+jobs:
+  send-message:
+    runs-on: ubuntu-latest
+    name: Send message via Matrix
+    steps:
+      - name: checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+      - run: |
+          ALL_MSGS=""
+          for i in ${{ join(github.event.commits.*.id, ' ') }}; do
+            MSG=$(git --no-pager show -s --format='%h <b>%an</b>: %s' $i)
+            ALL_MSGS="$ALL_MSGS$MSG<br>"
+          done
+          echo "::set-output name=COMMIT_MESSAGE::$ALL_MSGS"
+        id: commit-message
+      - uses: narodnik/matrix-action@main
+        with:
+          server: 'matrix.dark.fi'
+          room-id: '!MODZOZydPqCRdulXmR:dark.fi'
+          #access_token: ${{ secrets.MATRIX_TOKEN }}
+          status: 'OK'
+          user: 'narodnik'
+          password: ${{ secrets.MATRIX_PASSWORD }}
+          message: '${{ steps.commit-message.outputs.COMMIT_MESSAGE }}'
+          #- name: Send message to test channel
+          #  id: matrix-chat-message
+          #  uses: fadenb/matrix-chat-message@v0.0.6
+          #  with:
+          #    homeserver: 'dark.fi'
+          #    token: ${{ secrets.MATRIX_TOKEN }}
+          #    channel: '!MODZOZydPqCRdulXmR:dark.fi'
+          #    message: |
+          #      This is an *example message* using **markdown** for formatting.\
+          #      Use a `\` character at the end of a line to cause a linebreak (the whole message is treated as markdown).\
+          #      You can use variables like ${{ github.sha }} anywhere.
+

run_network.sh

@@ -0,0 +1,45 @@
+#!/bin/bash
+
+# Run this script then
+#   python scripts/monitor-p2p.py
+# to view the network topology
+
+declare -a arr=(
+    # Seed node
+    "cargo run --bin dfi -- -r 8999 --accept 127.0.0.1:9999 --log /tmp/darkfi/seed.log"
+    # Server with no outgoing connections
+    "cargo run --bin dfi -- -r 9000 --accept 127.0.0.1:10001 --seeds 127.0.0.1:9999 --log /tmp/darkfi/server.log"
+    # Server/client with 2 outgoing connections
+    "cargo run --bin dfi -- -r 9005 --accept 127.0.0.1:10002 --seeds 127.0.0.1:9999 --log /tmp/darkfi/server1.log --slots 3"
+    # Server/client with 2 outgoing connections
+    "cargo run --bin dfi -- -r 9006 --accept 127.0.0.1:10003 --seeds 127.0.0.1:9999 --log /tmp/darkfi/server2.log --slots 3"
+    # Server/client with 2 outgoing connections
+    "cargo run --bin dfi -- -r 9007 --accept 127.0.0.1:10004 --seeds 127.0.0.1:9999 --log /tmp/darkfi/server3.log --slots 3"
+    # Server/client with 2 outgoing connections
+    "cargo run --bin dfi -- -r 9008 --accept 127.0.0.1:10005 --seeds 127.0.0.1:9999 --log /tmp/darkfi/server4.log --slots 3"
+    # Client with 1 outgoing connection
+    "cargo run --bin dfi -- -r 9002 --seeds 127.0.0.1:9999 --slots 4 --log /tmp/darkfi/client.log"
+    # Client with 1 outgoing connection
+    "cargo run --bin dfi -- -r 9003 --seeds 127.0.0.1:9999 --slots 4 --log /tmp/darkfi/client1.log"
+    # Client with 1 outgoing connection
+    "cargo run --bin dfi -- -r 9004 --seeds 127.0.0.1:9999 --slots 4 --log /tmp/darkfi/client2.log"
+)
+
+mkdir -p /tmp/darkfi/
+
+for cmd in "${arr[@]}"; do {
+  echo "Process \"$cmd\" started";
+  RUST_BACKTRACE=1 $cmd & pid=$!
+  PID_LIST+=" $pid";
+  sleep 2;
+} done
+
+trap "kill $PID_LIST" SIGINT
+
+echo "Parallel processes have started";
+
+wait $PID_LIST
+
+echo
+echo "All processes have completed";
+

scripts/monitor-p2p.py

@@ -0,0 +1,159 @@
+import asyncio
+from tabulate import tabulate
+from copy import deepcopy
+import re
+import os
+import sys
+import time
+
+lock = asyncio.Lock()
+logs_path = "/tmp/darkfi/"
+
+node_info = {
+}
+
+ping_times = {
+}
+
+def debug(line):
+    #print(line)
+    pass
+
+def process(info, line):
+    regex_listen = re.compile(
+        ".* Listening on (\d+[.]\d+[.]\d+[.]\d+:\d+)")
+    regex_inbound_connect = re.compile(
+        ".* Connected inbound \[(\d+[.]\d+[.]\d+[.]\d+:\d+)\]")
+    regex_outbound_slots = re.compile(
+        ".* Starting (\d+) outbound connection slots.")
+    regex_outbound_connect = re.compile(
+        ".* #(\d+) connected to outbound \[(\d+[.]\d+[.]\d+[.]\d+:\d+)\]")
+    regex_channel_disconnected = re.compile(
+        ".* Channel (\d+[.]\d+[.]\d+[.]\d+:\d+) disconnected")
+    regex_pong_recv = re.compile(
+        ".* Received Pong message (\d+)ms from \[(\d+[.]\d+[.]\d+[.]\d+:\d+)\]")
+
+    if "net: P2p::start() [BEGIN]" in line:
+        info["status"] = "p2p-start"
+    elif "net: SeedSession::start() [START]" in line:
+        info["status"] = "seed-start"
+    elif "net: SeedSession::start() [END]" in line:
+        info["status"] = "seed-done"
+    elif "net: P2p::start() [END]" in line:
+        info["status"] = "p2p-done"
+    elif "net: P2p::run() [BEGIN]" in line:
+        info["status"] = "p2p-run"
+    elif "Not configured for accepting incoming connections." in line:
+        info["inbounds"] = ["Disabled"]
+    elif (match := regex_listen.match(line)) is not None:
+        address = match.group(1)
+        info["listen"] = address
+    elif (match := regex_inbound_connect.match(line)) is not None:
+        address = match.group(1)
+        info["inbounds"].append(address)
+    elif (match := regex_outbound_slots.match(line)) is not None:
+        slots = match.group(1)
+        info["outbounds"] = ["None" for _ in range(int(slots))]
+    elif (match := regex_outbound_connect.match(line)) is not None:
+        slot = match.group(1)
+        address = match.group(2)
+        info["outbounds"][int(slot)] = address
+    elif (match := regex_channel_disconnected.match(line)) is not None:
+        address = match.group(1)
+        try:
+            info["inbounds"].remove(address)
+        except ValueError:
+            pass
+        try:
+            idx = info["outbounds"].index(address)
+            info["outbounds"][idx] = "None"
+        except ValueError:
+            pass
+    elif (match := regex_pong_recv.match(line)) is not None:
+        ping_time = match.group(1)
+        address = match.group(2)
+        ping_times[address] = ping_time
+
+async def scanner(filename):
+    global table_data
+
+    async with lock:
+        node_info[filename] = {
+            "status": "none",
+            "inbounds": [],
+            "outbounds": [],
+        }
+        info = node_info[filename]
+
+    with open(logs_path + filename) as fileh:
+        while True:
+            line = fileh.readline()
+            if line:
+                debug("R: " + filename + ": " + line[:-1])
+                async with lock:
+                    process(info, line)
+            else:
+                await asyncio.sleep(0.5)
+
+def clear_lines(n):
+    for i in range(n):
+        sys.stdout.write('\033[F')
+
+def get_ping(addr):
+    ping_time = "none"
+    if addr in ping_times:
+        ping_time = str(ping_times[addr]) + " ms"
+    return ping_time
+
+def table_format(ninfo):
+    table_data = []
+    for filename, info in ninfo.items():
+        table_data.append([filename, "", ""])
+        table_data.append(["", "status", info["status"]])
+
+        if "listen" in info:
+            table_data.append(["", "listen", info["listen"]])
+
+        inbounds = info["inbounds"]
+        if inbounds:
+            table_data.append(["", "inbounds", inbounds[0],
+                               get_ping(inbounds[0])])
+
+            for inbound in inbounds[1:]:
+                table_data.append(["", "", inbound, get_ping(inbound)])
+
+        outbounds = info["outbounds"]
+        if outbounds:
+            table_data.append(["", "outbounds", outbounds[0],
+                               get_ping(outbounds[0])])
+
+            for outbound in outbounds[1:]:
+                table_data.append(["", "", outbound, get_ping(outbound)])
+
+    headers = ["Name", "Attribute", "Value", "Ping Times"]
+    return headers, table_data
+
+async def refresh_table(tick=1):
+    for filename in os.listdir(logs_path):
+        asyncio.create_task(scanner(filename))
+
+    previous_lines = 0
+
+    while True:
+        clear_lines(previous_lines)
+
+        async with lock:
+            ninfo = deepcopy(node_info)
+        headers, table_data = table_format(ninfo)
+        lines = tabulate(table_data, headers=headers).split("\n")
+        debug("-------------------")
+        for line in lines:
+            print('\x1b[2K\r', end="")
+            print(line)
+
+        previous_lines = len(lines)
+
+        await asyncio.sleep(1)
+
+asyncio.run(refresh_table())
+

src/bin/dfi.rs

@@ -5,6 +5,7 @@ use async_native_tls::TlsAcceptor;
 use async_std::sync::Mutex;
 use easy_parallel::Parallel;
 use http_types::{Request, Response, StatusCode};
+use log::*;
 use serde_json::json;
 use smol::Async;
 use std::net::SocketAddr;
@@ -92,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?;
 
@@ -152,6 +153,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     *rpc.started.lock().await = true;
 
     p2p.clone().start(executor.clone()).await?;
+
     p2p.run(executor).await?;
 
     rpc.wait_for_quit().await?;
@@ -269,7 +271,6 @@ impl ProgramOptions {
             (@arg CONNECTS: -c --connect ... "Manual connections")
             (@arg CONNECT_SLOTS: --slots +takes_value "Connection slots")
             (@arg LOG_PATH: --log +takes_value "Logfile path")
-            (@arg DISABLE_SEED: -D --disable_seed "Disable seed process")
             (@arg RPC_PORT: -r --rpc +takes_value "RPC port")
         )
         .get_matches();
@@ -309,12 +310,6 @@ impl ProgramOptions {
             .to_path_buf(),
         );
 
-        let skip_seed_sync = if app.is_present("DISABLE_SEED") {
-            true
-        } else {
-            false
-        };
-
         let rpc_port = if let Some(rpc_port) = app.value_of("RPC_PORT") {
             rpc_port.parse()?
         } else {
@@ -325,13 +320,13 @@ impl ProgramOptions {
             network_settings: net::Settings {
                 inbound: accept_addr,
                 outbound_connections: connection_slots,
+                seed_query_timeout_seconds: 8,
                 connect_timeout_seconds: 10,
                 channel_handshake_seconds: 4,
                 channel_heartbeat_seconds: 10,
                 external_addr: accept_addr,
                 peers: manual_connects,
                 seeds: seed_addrs,
-                skip_seed_sync,
             },
             log_path,
             rpc_port,

src/bin/mimc.rs

@@ -1,5 +1,5 @@
 use bls12_381::Scalar;
-use ff::{Field};
+use ff::Field;
 use sapvi::{Decodable, ZKContract};
 use std::fs::File;
 use std::ops::{Add, AddAssign, MulAssign, Neg, SubAssign};

src/bin/mimc_constants.rs

@@ -325,5 +325,4 @@ pub fn mimc_constants() -> Vec<&'static str> {
     ]
 }
 
-fn main() {
-}
+fn main() {}

src/bls_extensions.rs

@@ -5,77 +5,77 @@ use crate::error::{Error, Result};
 use crate::serial::{Decodable, Encodable, ReadExt, WriteExt};
 
 macro_rules! from_slice {
-($data:expr, $len:literal) => {{
-let mut array = [0; $len];
-// panics if not enough data
-let bytes = &$data[..array.len()];
-array.copy_from_slice(bytes);
-array
-}};
+    ($data:expr, $len:literal) => {{
+        let mut array = [0; $len];
+        // panics if not enough data
+        let bytes = &$data[..array.len()];
+        array.copy_from_slice(bytes);
+        array
+    }};
 }
 
 pub trait BlsStringConversion {
-fn to_string(&self) -> String;
-fn from_string(object: &str) -> Self;
+    fn to_string(&self) -> String;
+    fn from_string(object: &str) -> Self;
 }
 
 impl BlsStringConversion for bls::Scalar {
-fn to_string(&self) -> String {
-let mut bytes = self.to_bytes();
-bytes.reverse();
-hex::encode(bytes)
-}
-fn from_string(object: &str) -> Self {
-let mut bytes = from_slice!(&hex::decode(object).unwrap(), 32);
-bytes.reverse();
-bls::Scalar::from_bytes(&bytes).unwrap()
-}
+    fn to_string(&self) -> String {
+        let mut bytes = self.to_bytes();
+        bytes.reverse();
+        hex::encode(bytes)
+    }
+    fn from_string(object: &str) -> Self {
+        let mut bytes = from_slice!(&hex::decode(object).unwrap(), 32);
+        bytes.reverse();
+        bls::Scalar::from_bytes(&bytes).unwrap()
+    }
 }
 
 macro_rules! serialization_bls {
-($type:ty, $to_x:ident, $from_x:ident, $size:literal) => {
-impl Encodable for $type {
-fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-let data = self.$to_x();
-assert_eq!(data.len(), $size);
-s.write_slice(&data)?;
-Ok(data.len())
-}
-}
+    ($type:ty, $to_x:ident, $from_x:ident, $size:literal) => {
+        impl Encodable for $type {
+            fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+                let data = self.$to_x();
+                assert_eq!(data.len(), $size);
+                s.write_slice(&data)?;
+                Ok(data.len())
+            }
+        }
 
-impl Decodable for $type {
-fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-let mut slice = [0u8; $size];
-d.read_slice(&mut slice)?;
-let result = Self::$from_x(&slice);
-if bool::from(result.is_none()) {
-return Err(Error::ParseFailed("$t conversion from slice failed"));
-}
-Ok(result.unwrap())
-}
-}
-};
+        impl Decodable for $type {
+            fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+                let mut slice = [0u8; $size];
+                d.read_slice(&mut slice)?;
+                let result = Self::$from_x(&slice);
+                if bool::from(result.is_none()) {
+                    return Err(Error::ParseFailed("$t conversion from slice failed"));
+                }
+                Ok(result.unwrap())
+            }
+        }
+    };
 }
 
 serialization_bls!(bls::Scalar, to_bytes, from_bytes, 32);
 
 macro_rules! make_serialize_deserialize_test {
-($name:ident, $type:ty, $default_func:ident) => {
-#[test]
-fn $name() {
-let point = <$type>::$default_func();
+    ($name:ident, $type:ty, $default_func:ident) => {
+        #[test]
+        fn $name() {
+            let point = <$type>::$default_func();
 
-let mut data: Vec<u8> = vec![];
-let result = point.encode(&mut data);
-assert!(result.is_ok());
+            let mut data: Vec<u8> = vec![];
+            let result = point.encode(&mut data);
+            assert!(result.is_ok());
 
-let point2 = <$type>::decode(&data[..]);
-assert!(point2.is_ok());
-let point2 = point2.unwrap();
+            let point2 = <$type>::decode(&data[..]);
+            assert!(point2.is_ok());
+            let point2 = point2.unwrap();
 
-assert_eq!(point, point2);
-}
-};
+            assert_eq!(point, point2);
+        }
+    };
 }
 
 make_serialize_deserialize_test!(serial_test_scalar, bls::Scalar, zero);

src/error.rs

@@ -39,6 +39,7 @@ pub enum Error {
     ChannelStopped,
     ChannelTimeout,
     ServiceStopped,
+    Utf8Error,
 }
 
 impl std::error::Error for Error {}
@@ -80,6 +81,7 @@ impl fmt::Display for Error {
             Error::ChannelStopped => f.write_str("Channel stopped"),
             Error::ChannelTimeout => f.write_str("Channel timed out"),
             Error::ServiceStopped => f.write_str("Service stopped"),
+            Error::Utf8Error => f.write_str("Malformed UTF8"),
         }
     }
 }
@@ -138,3 +140,9 @@ impl From<NetError> for Error {
         }
     }
 }
+
+impl From<std::string::FromUtf8Error> for 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/acceptor.rs

@@ -4,7 +4,7 @@ use std::net::{SocketAddr, TcpListener};
 use std::sync::Arc;
 
 use crate::net::error::{NetError, NetResult};
-use crate::net::{Channel, ChannelPtr, SettingsPtr};
+use crate::net::{Channel, ChannelPtr};
 use crate::system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription};
 
 pub type AcceptorPtr = Arc<Acceptor>;
@@ -12,15 +12,13 @@ pub type AcceptorPtr = Arc<Acceptor>;
 pub struct Acceptor {
     channel_subscriber: SubscriberPtr<NetResult<ChannelPtr>>,
     task: StoppableTaskPtr,
-    settings: SettingsPtr,
 }
 
 impl Acceptor {
-    pub fn new(settings: SettingsPtr) -> Arc<Self> {
+    pub fn new() -> Arc<Self> {
         Arc::new(Self {
             channel_subscriber: Subscriber::new(),
             task: StoppableTask::new(),
-            settings,
         })
     }
 
@@ -48,14 +46,14 @@ impl Acceptor {
 
     fn setup(accept_addr: SocketAddr) -> NetResult<Async<TcpListener>> {
         let listener = match Async::<TcpListener>::bind(accept_addr) {
-            Ok(l) => l,
+            Ok(listener) => listener,
             Err(err) => {
                 error!("Bind listener failed: {}", err);
                 return Err(NetError::OperationFailed);
             }
         };
         let local_addr = match listener.get_ref().local_addr() {
-            Ok(a) => a,
+            Ok(addr) => addr,
             Err(err) => {
                 error!("Failed to get local address: {}", err);
                 return Err(NetError::OperationFailed);
@@ -78,8 +76,7 @@ impl Acceptor {
     async fn run_accept_loop(self: Arc<Self>, listener: Async<TcpListener>) -> NetResult<()> {
         loop {
             let channel = self.tick_accept(&listener).await?;
-            let channel_result = Arc::new(Ok(channel));
-            self.channel_subscriber.notify(channel_result).await;
+            self.channel_subscriber.notify(Ok(channel)).await;
         }
     }
 
@@ -88,7 +85,7 @@ impl Acceptor {
             Ok(()) => panic!("Acceptor task should never complete without error status"),
             Err(err) => {
                 // Send this error to all channel subscribers
-                let result = Arc::new(Err(err));
+                let result = Err(err);
                 self.channel_subscriber.notify(result).await;
             }
         }
@@ -104,7 +101,7 @@ impl Acceptor {
         };
         info!("Accepted client: {}", peer_addr);
 
-        let channel = Channel::new(stream, peer_addr, self.settings.clone());
+        let channel = Channel::new(stream, peer_addr).await;
         Ok(channel)
     }
 }

src/net/channel.rs

@@ -11,11 +11,8 @@ use std::sync::Arc;
 
 use crate::error;
 use crate::net::error::{NetError, NetResult};
-use crate::net::message_subscriber::{
-    MessageSubscriber, MessageSubscriberPtr, MessageSubscription,
-};
+use crate::net::message_subscriber::{MessageSubscription, MessageSubsystem};
 use crate::net::messages;
-use crate::net::settings::SettingsPtr;
 use crate::system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription};
 
 pub type ChannelPtr = Arc<Channel>;
@@ -24,27 +21,32 @@ pub struct Channel {
     reader: Mutex<ReadHalf<Async<TcpStream>>>,
     writer: Mutex<WriteHalf<Async<TcpStream>>>,
     address: SocketAddr,
-    message_subscriber: MessageSubscriberPtr,
+    message_subsystem: MessageSubsystem,
     stop_subscriber: SubscriberPtr<NetError>,
     receive_task: StoppableTaskPtr,
     stopped: AtomicBool,
-    settings: SettingsPtr,
 }
 
 impl Channel {
-    pub fn new(stream: Async<TcpStream>, address: SocketAddr, settings: SettingsPtr) -> Arc<Self> {
+    pub async fn new(
+        stream: Async<TcpStream>,
+        address: SocketAddr,
+    ) -> Arc<Self> {
         let (reader, writer) = stream.split();
         let reader = Mutex::new(reader);
         let writer = Mutex::new(writer);
+
+        let message_subsystem = MessageSubsystem::new();
+        Self::setup_dispatchers(&message_subsystem).await;
+
         Arc::new(Self {
             reader,
             writer,
             address,
-            message_subscriber: MessageSubscriber::new(),
+            message_subsystem,
             stop_subscriber: Subscriber::new(),
             receive_task: StoppableTask::new(),
             stopped: AtomicBool::new(false),
-            settings,
         })
     }
 
@@ -52,7 +54,7 @@ impl Channel {
         debug!(target: "net", "Channel::start() [START, address={}]", self.address());
         let self2 = self.clone();
         self.receive_task.clone().start(
-            self.clone().receive_loop(),
+            self.clone().main_receive_loop(),
             // Ignore stop handler
             |result| self2.handle_stop(result),
             NetError::ServiceStopped,
@@ -65,9 +67,9 @@ impl Channel {
         debug!(target: "net", "Channel::stop() [START, address={}]", self.address());
         assert_eq!(self.stopped.load(Ordering::Relaxed), false);
         self.stopped.store(false, Ordering::Relaxed);
-        let stop_err = Arc::new(NetError::ChannelStopped);
-        self.stop_subscriber.notify(stop_err).await;
+        self.stop_subscriber.notify(NetError::ChannelStopped).await;
         self.receive_task.stop().await;
+        self.message_subsystem.trigger_error(NetError::ChannelStopped).await;
         debug!(target: "net", "Channel::stop() [END, address={}]", self.address());
     }
 
@@ -86,11 +88,10 @@ impl Channel {
         sub
     }
 
-    pub async fn send(self: Arc<Self>, message: messages::Message) -> NetResult<()> {
-        let packet_type = message.packet_type();
+    pub async fn send<M: messages::Message>(&self, message: M) -> NetResult<()> {
         debug!(target: "net",
-            "Channel::send() [START, pkt_type={:?}, address={}]",
-            packet_type,
+            "Channel::send() [START, command={:?}, address={}]",
+            M::name(),
             self.address()
         );
         if self.stopped.load(Ordering::Relaxed) {
@@ -98,7 +99,7 @@ impl Channel {
         }
 
         // Catch failure and stop channel, return a net error
-        let result = match messages::send_message(&mut *self.writer.lock().await, message).await {
+        let result = match self.send_message(message).await {
             Ok(()) => Ok(()),
             Err(err) => {
                 error!("Channel send error for [{}]: {}", self.address(), err);
@@ -107,26 +108,35 @@ impl Channel {
             }
         };
         debug!(target: "net",
-            "Channel::send() [END, pkt_type={:?}, address={}]",
-            packet_type,
+            "Channel::send() [END, command={:?}, address={}]",
+            M::name(),
             self.address()
         );
         result
     }
 
-    pub async fn subscribe_msg(
-        self: Arc<Self>,
-        packet_type: messages::PacketType,
-    ) -> MessageSubscription {
+    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: messages::Message>(&self) -> NetResult<MessageSubscription<M>> {
         debug!(target: "net",
-            "Channel::subscribe_msg() [START, pkt_type={:?}, address={}]",
-            packet_type,
+            "Channel::subscribe_msg() [START, command={:?}, address={}]",
+            M::name(),
             self.address()
         );
-        let sub = self.message_subscriber.clone().subscribe(packet_type).await;
+        let sub = self.message_subsystem.subscribe::<M>().await;
         debug!(target: "net",
-            "Channel::subscribe_msg() [END, pkt_type={:?}, address={}]",
-            packet_type,
+            "Channel::subscribe_msg() [END, command={:?}, address={}]",
+            M::name(),
             self.address()
         );
         sub
@@ -143,17 +153,42 @@ impl Channel {
         }
     }
 
-    async fn receive_loop(self: Arc<Self>) -> NetResult<()> {
+    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;
+    }
+
+    pub fn get_message_subsystem(&self) -> &MessageSubsystem {
+        &self.message_subsystem
+    }
+
+    async fn main_receive_loop(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net",
             "Channel::receive_loop() [START, address={}]",
             self.address()
         );
+
         let reader = &mut *self.reader.lock().await;
 
         loop {
-            let message_result = messages::receive_message(reader).await;
-            let message = match message_result {
-                Ok(message) => Arc::new(message),
+            let packet = match messages::read_packet(reader).await {
+                Ok(packet) => packet,
                 Err(err) => {
                     if Self::is_eof_error(&err) {
                         info!("Channel {} disconnected", self.address());
@@ -170,7 +205,9 @@ impl Channel {
             };
 
             // Send result to our subscribers
-            self.message_subscriber.notify(Ok(message)).await;
+            self.message_subsystem
+                .notify(&packet.command, packet.payload)
+                .await;
         }
     }
 
@@ -180,8 +217,7 @@ impl Channel {
             Ok(()) => panic!("Channel task should never complete without error status"),
             Err(err) => {
                 // Send this error to all channel subscribers
-                let result = Err(err);
-                self.message_subscriber.notify(result).await;
+                self.message_subsystem.trigger_error(err).await;
             }
         }
         debug!(target: "net", "Channel::handle_stop() [END, address={}]", self.address());

src/net/connector.rs

@@ -19,7 +19,7 @@ impl Connector {
         futures::select! {
             stream_result = Async::<TcpStream>::connect(hostaddr).fuse() => {
                 match stream_result {
-                    Ok(stream) => Ok(Channel::new(stream, hostaddr, self.settings.clone())),
+                    Ok(stream) => Ok(Channel::new(stream, hostaddr).await),
                     Err(_) => Err(NetError::ConnectFailed)
                 }
             }

src/net/hosts.rs

@@ -2,26 +2,30 @@ use async_std::sync::Mutex;
 use rand::seq::SliceRandom;
 use std::net::SocketAddr;
 use std::sync::Arc;
-
-use crate::net::SettingsPtr;
+use std::collections::HashSet;
 
 pub type HostsPtr = Arc<Hosts>;
 
 pub struct Hosts {
     addrs: Mutex<Vec<SocketAddr>>,
-    settings: SettingsPtr,
 }
 
 impl Hosts {
-    pub fn new(settings: SettingsPtr) -> Arc<Self> {
+    pub fn new() -> Arc<Self> {
         Arc::new(Self {
             addrs: Mutex::new(Vec::new()),
-            settings,
         })
     }
 
+    async fn contains(&self, addrs: &Vec<SocketAddr>) -> bool {
+        let a_set: HashSet<_> = addrs.iter().copied().collect();
+        self.addrs.lock().await.iter().any(|item| a_set.contains(item))
+    }
+
     pub async fn store(&self, addrs: Vec<SocketAddr>) {
-        self.addrs.lock().await.extend(addrs)
+        if !self.contains(&addrs).await {
+            self.addrs.lock().await.extend(addrs)
+        }
     }
 
     pub async fn load_single(&self) -> Option<SocketAddr> {
@@ -35,4 +39,8 @@ impl Hosts {
     pub async fn load_all(&self) -> Vec<SocketAddr> {
         self.addrs.lock().await.clone()
     }
+
+    pub async fn is_empty(&self) -> bool {
+        self.addrs.lock().await.is_empty()
+    }
 }

src/net/message_subscriber.rs

@@ -1,65 +1,33 @@
 use async_std::sync::Mutex;
+use async_trait::async_trait;
+use log::*;
 use rand::Rng;
+use std::any::Any;
 use std::collections::HashMap;
+use std::io;
+use std::io::Cursor;
 use std::sync::Arc;
 
-use crate::net::error::NetResult;
-use crate::net::messages::{Message, PacketType};
+use crate::error::Result;
+use crate::net::error::{NetError, NetResult};
+use crate::net::messages::Message;
+use crate::serial::{Decodable, Encodable};
 
-pub type MessageSubscriberPtr = Arc<MessageSubscriber>;
-
-pub type MessageResult = NetResult<Arc<Message>>;
 pub type MessageSubscriptionID = u64;
+type MessageResult<M> = NetResult<Arc<M>>;
 
-macro_rules! receive_message {
-    ($sub:expr, $message_type:path) => {{
-        let wrapped_message = owning_ref::OwningRef::new($sub.receive().await?);
-
-        wrapped_message.map(|msg| match msg {
-            $message_type(msg_detail) => msg_detail,
-            _ => {
-                panic!("Filter for receive sub invalid!");
-            }
-        })
-    }};
-}
-
-pub struct MessageSubscription {
+pub struct MessageSubscription<M: Message> {
     id: MessageSubscriptionID,
-    filter: PacketType,
-    recv_queue: async_channel::Receiver<MessageResult>,
-    parent: Arc<MessageSubscriber>,
+    recv_queue: async_channel::Receiver<MessageResult<M>>,
+    parent: Arc<MessageDispatcher<M>>,
 }
 
-impl MessageSubscription {
-    fn is_relevant_message(&self, message_result: &MessageResult) -> bool {
-        match message_result {
-            Ok(message) => {
-                let packet_type = message.packet_type();
-
-                // Apply the filter
-                packet_type == self.filter
-            }
-            Err(_) => {
-                // Propagate all errors
-                true
-            }
-        }
-    }
-
-    pub async fn receive(&self) -> MessageResult {
-        loop {
-            let message_result = self.recv_queue.recv().await;
-
-            match message_result {
-                Ok(message_result) => {
-                    if self.clone().is_relevant_message(&message_result) {
-                        return message_result;
-                    }
-                }
-                Err(err) => {
-                    panic!("MessageSubscription::receive() recv_queue failed! {}", err);
-                }
+impl<M: Message> MessageSubscription<M> {
+    pub async fn receive(&self) -> MessageResult<M> {
+        match self.recv_queue.recv().await {
+            Ok(message) => message,
+            Err(err) => {
+                panic!("MessageSubscription::receive() recv_queue failed! {}", err);
             }
         }
     }
@@ -70,15 +38,24 @@ impl MessageSubscription {
     }
 }
 
-pub struct MessageSubscriber {
-    subs: Mutex<HashMap<MessageSubscriptionID, async_channel::Sender<MessageResult>>>,
+#[async_trait]
+trait MessageDispatcherInterface: Send + Sync {
+    async fn trigger(&self, payload: Vec<u8>);
+
+    async fn trigger_error(&self, err: NetError);
+
+    fn as_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync>;
+}
+
+struct MessageDispatcher<M: Message> {
+    subs: Mutex<HashMap<MessageSubscriptionID, async_channel::Sender<MessageResult<M>>>>,
 }
 
-impl MessageSubscriber {
-    pub fn new() -> Arc<Self> {
-        Arc::new(Self {
+impl<M: Message> MessageDispatcher<M> {
+    fn new() -> Self {
+        MessageDispatcher {
             subs: Mutex::new(HashMap::new()),
-        })
+        }
     }
 
     pub fn random_id() -> MessageSubscriptionID {
@@ -86,30 +63,33 @@ impl MessageSubscriber {
         rng.gen()
     }
 
-    pub async fn subscribe(self: Arc<Self>, packet_type: PacketType) -> MessageSubscription {
+    pub async fn subscribe(self: Arc<Self>) -> MessageSubscription<M> {
         let (sender, recvr) = async_channel::unbounded();
-
         let sub_id = Self::random_id();
-
         self.subs.lock().await.insert(sub_id, sender);
 
         MessageSubscription {
             id: sub_id,
-            filter: packet_type,
             recv_queue: recvr,
-            parent: self.clone(),
+            parent: self,
         }
     }
 
-    async fn unsubscribe(self: Arc<Self>, sub_id: MessageSubscriptionID) {
+    async fn unsubscribe(&self, sub_id: MessageSubscriptionID) {
         self.subs.lock().await.remove(&sub_id);
     }
 
-    pub async fn notify(&self, message_result: NetResult<Arc<Message>>) {
+    async fn trigger_all(&self, message: MessageResult<M>) {
+        debug!(
+            "MessageDispatcher<M={}>::trigger_all({}) [START, subs={}]",
+            M::name(),
+            if message.is_ok() { "msg" } else { "err" },
+            self.subs.lock().await.len()
+        );
         let mut garbage_ids = Vec::new();
 
         for (sub_id, sub) in &*self.subs.lock().await {
-            match sub.send(message_result.clone()).await {
+            match sub.send(message.clone()).await {
                 Ok(()) => {}
                 Err(_err) => {
                     // Automatically clean out closed channels
@@ -120,6 +100,13 @@ impl MessageSubscriber {
         }
 
         self.collect_garbage(garbage_ids).await;
+
+        debug!(
+            "MessageDispatcher<M={}>::trigger_all({}) [END, subs={}]",
+            M::name(),
+            if message.is_ok() { "msg" } else { "err" },
+            self.subs.lock().await.len()
+        );
     }
 
     async fn collect_garbage(&self, ids: Vec<MessageSubscriptionID>) {
@@ -129,3 +116,169 @@ impl MessageSubscriber {
         }
     }
 }
+
+#[async_trait]
+impl<M: Message> MessageDispatcherInterface for MessageDispatcher<M> {
+    async fn trigger(&self, payload: Vec<u8>) {
+        // deserialize data into type
+        // send down the pipes
+        let cursor = Cursor::new(payload);
+        match M::decode(cursor) {
+            Ok(message) => {
+                let message = Ok(Arc::new(message));
+                self.trigger_all(message).await
+            }
+            Err(err) => {
+                error!("Unable to decode data. Dropping...: {}", err);
+            }
+        }
+    }
+
+    async fn trigger_error(&self, err: NetError) {
+        self.trigger_all(Err(err)).await;
+    }
+
+    fn as_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
+        self
+    }
+}
+
+// NOTE: this class is a more general version of system::Subscriber which can dispatch
+// multiple different type of registered types to sub-dispatchers
+pub struct MessageSubsystem {
+    dispatchers: Mutex<HashMap<&'static str, Arc<dyn MessageDispatcherInterface>>>,
+}
+
+impl MessageSubsystem {
+    pub fn new() -> Self {
+        MessageSubsystem {
+            dispatchers: Mutex::new(HashMap::new()),
+        }
+    }
+
+    pub async fn add_dispatch<M: Message>(&self) {
+        self.dispatchers
+            .lock()
+            .await
+            .insert(M::name(), Arc::new(MessageDispatcher::<M>::new()));
+    }
+
+    pub async fn subscribe<M: Message>(&self) -> NetResult<MessageSubscription<M>> {
+        let dispatcher = self.dispatchers.lock().await.get(M::name()).cloned();
+
+        let sub = match dispatcher {
+            Some(dispatcher) => {
+                let dispatcher: Arc<MessageDispatcher<M>> = dispatcher
+                    .as_any()
+                    .downcast::<MessageDispatcher<M>>()
+                    .expect("Multiple messages registered with different names");
+
+                dispatcher.subscribe().await
+            }
+            None => {
+                // normall return failure here
+                // for now panic
+                return Err(NetError::OperationFailed);
+            }
+        };
+
+        Ok(sub)
+    }
+
+    pub async fn notify(&self, command: &str, payload: Vec<u8>) {
+        let dispatcher = self.dispatchers.lock().await.get(command).cloned();
+
+        match dispatcher {
+            Some(dispatcher) => {
+                dispatcher.trigger(payload).await;
+            }
+            None => {
+                warn!(
+                    "MessageSubsystem::notify(\"{}\", payload) did not find a dispatcher",
+                    command
+                );
+            }
+        }
+    }
+
+    pub async fn trigger_error(&self, err: NetError) {
+        // TODO: this could be parallelized
+        for dispatcher in self.dispatchers.lock().await.values() {
+            dispatcher.trigger_error(err).await;
+        }
+    }
+}
+
+// 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
+async fn _do_message_subscriber_test() {
+    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();
+    subsystem.add_dispatch::<MyVersionMessage>().await;
+
+    // subscribe
+    //   1. get dispatcher
+    //   2. cast to specific type
+    //   3. do sub, return sub
+    let sub = subsystem.subscribe::<MyVersionMessage>().await.unwrap();
+
+    let msg = MyVersionMessage { x: 110 };
+    let mut payload = Vec::new();
+    msg.encode(&mut payload).unwrap();
+
+    // receive message and publish
+    //   1. based on string, lookup relevant dispatcher interface
+    //   2. publish data there
+    subsystem.notify("verver", payload).await;
+
+    // receive
+    //    1. do a get easy
+    let msg2 = sub.receive().await.unwrap();
+    assert_eq!(msg2.x, 110);
+    println!("{}", msg2.x);
+
+    subsystem.trigger_error(NetError::ChannelStopped).await;
+
+    let msg2 = sub.receive().await;
+    assert!(msg2.is_err());
+
+    sub.unsubscribe().await;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_message_subscriber() {
+        smol::block_on(_do_message_subscriber_test());
+    }
+}
+

src/net/messages.rs

@@ -1,51 +1,15 @@
 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 type Ciphertext = Vec<u8>;
-pub type CiphertextHash = [u8; 32];
-
-// Packets and Message because Rust doesn't allow value
-// aliasing from ADL type enums (which Message uses).
-#[derive(IntoPrimitive, TryFromPrimitive, Copy, Clone, PartialEq, Eq, Hash, Debug)]
-#[repr(u8)]
-pub enum PacketType {
-    Ping = 1,
-    Pong = 2,
-    GetAddrs = 3,
-    Addrs = 4,
-    Inv = 5,
-    GetSlabs = 6,
-    Slab = 7,
-    Version = 8,
-    Verack = 9,
-}
-
-pub enum Message {
-    Ping(PingMessage),
-    Pong(PongMessage),
-    GetAddrs(GetAddrsMessage),
-    Addrs(AddrsMessage),
-    Inv(InvMessage),
-    GetSlabs(GetSlabsMessage),
-    Slab(SlabMessage),
-    Version(VersionMessage),
-    Verack(VerackMessage),
+pub trait Message: 'static + Encodable + Decodable + Send + Sync {
+    fn name() -> &'static str;
 }
 
 pub struct PingMessage {
@@ -58,20 +22,6 @@ pub struct PongMessage {
 
 pub struct GetAddrsMessage {}
 
-pub struct GetSlabsMessage {
-    pub slabs_hash: Vec<[u8; 32]>,
-}
-
-#[derive(Clone)]
-pub struct SlabMessage {
-    pub nonce: [u8; 12],
-    pub ciphertext: Ciphertext,
-}
-
-pub struct InvMessage {
-    pub slabs_hash: Vec<[u8; 32]>,
-}
-
 pub struct AddrsMessage {
     pub addrs: Vec<SocketAddr>,
 }
@@ -80,84 +30,70 @@ pub struct VersionMessage {}
 
 pub struct VerackMessage {}
 
-impl Encodable for PingMessage {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-        let mut len = 0;
-        len += self.nonce.encode(&mut s)?;
-        Ok(len)
+impl Message for PingMessage {
+    fn name() -> &'static str {
+        "ping"
     }
 }
 
-impl Decodable for PingMessage {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-        Ok(Self {
-            nonce: Decodable::decode(&mut d)?,
-        })
+impl Message for PongMessage {
+    fn name() -> &'static str {
+        "pong"
     }
 }
 
-impl Encodable for PongMessage {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-        let mut len = 0;
-        len += self.nonce.encode(&mut s)?;
-        Ok(len)
+impl Message for GetAddrsMessage {
+    fn name() -> &'static str {
+        "getaddr"
     }
 }
 
-impl Decodable for PongMessage {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-        Ok(Self {
-            nonce: Decodable::decode(&mut d)?,
-        })
+impl Message for AddrsMessage {
+    fn name() -> &'static str {
+        "addr"
     }
 }
 
-impl Encodable for GetSlabsMessage {
-    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-        let mut len = 0;
-        len += self.slabs_hash.encode(&mut s)?;
-        Ok(len)
+impl Message for VersionMessage {
+    fn name() -> &'static str {
+        "version"
     }
 }
 
-impl Decodable for GetSlabsMessage {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-        Ok(Self {
-            slabs_hash: Decodable::decode(&mut d)?,
-        })
+impl Message for VerackMessage {
+    fn name() -> &'static str {
+        "verack"
     }
 }
 
-impl Encodable for SlabMessage {
+impl Encodable for PingMessage {
     fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
         let mut len = 0;
         len += self.nonce.encode(&mut s)?;
-        len += self.ciphertext.encode(&mut s)?;
         Ok(len)
     }
 }
 
-impl Decodable for SlabMessage {
+impl Decodable for PingMessage {
     fn decode<D: io::Read>(mut d: D) -> Result<Self> {
         Ok(Self {
             nonce: Decodable::decode(&mut d)?,
-            ciphertext: Decodable::decode(&mut d)?,
         })
     }
 }
 
-impl Encodable for InvMessage {
+impl Encodable for PongMessage {
     fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
         let mut len = 0;
-        len += self.slabs_hash.encode(&mut s)?;
+        len += self.nonce.encode(&mut s)?;
         Ok(len)
     }
 }
 
-impl Decodable for InvMessage {
+impl Decodable for PongMessage {
     fn decode<D: io::Read>(mut d: D) -> Result<Self> {
         Ok(Self {
-            slabs_hash: Decodable::decode(&mut d)?,
+            nonce: Decodable::decode(&mut d)?,
         })
     }
 }
@@ -215,127 +151,10 @@ 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,
-        }
-    }
-
-    pub fn pack(&self) -> Result<Packet> {
-        match self {
-            Message::Ping(message) => {
-                let mut payload = Vec::new();
-                message.encode(&mut payload)?;
-                Ok(Packet {
-                    command: PacketType::Ping,
-                    payload,
-                })
-            }
-            Message::Pong(message) => {
-                let mut payload = Vec::new();
-                message.encode(&mut payload)?;
-                Ok(Packet {
-                    command: PacketType::Pong,
-                    payload,
-                })
-            }
-            Message::GetAddrs(message) => {
-                let mut payload = Vec::new();
-                message.encode(&mut payload)?;
-                Ok(Packet {
-                    command: PacketType::GetAddrs,
-                    payload,
-                })
-            }
-            Message::Addrs(message) => {
-                let mut payload = Vec::new();
-                message.encode(Cursor::new(&mut payload))?;
-                Ok(Packet {
-                    command: PacketType::Addrs,
-                    payload,
-                })
-            }
-            Message::Inv(message) => {
-                let payload = serialize(message);
-                Ok(Packet {
-                    command: PacketType::Inv,
-                    payload,
-                })
-            }
-            Message::GetSlabs(message) => {
-                let payload = serialize(message);
-                Ok(Packet {
-                    command: PacketType::GetSlabs,
-                    payload,
-                })
-            }
-            Message::Slab(message) => {
-                let payload = serialize(message);
-                Ok(Packet {
-                    command: PacketType::Slab,
-                    payload,
-                })
-            }
-            Message::Version(message) => {
-                let payload = serialize(message);
-                Ok(Packet {
-                    command: PacketType::Version,
-                    payload,
-                })
-            }
-            Message::Verack(message) => {
-                let payload = serialize(message);
-                Ok(Packet {
-                    command: PacketType::Verack,
-                    payload,
-                })
-            }
-        }
-    }
-
-    pub fn unpack(packet: Packet) -> Result<Self> {
-        let cursor = Cursor::new(packet.payload.clone());
-        match packet.command {
-            PacketType::Ping => Ok(Self::Ping(PingMessage::decode(cursor)?)),
-            PacketType::Pong => Ok(Self::Pong(PongMessage::decode(cursor)?)),
-            PacketType::GetAddrs => Ok(Self::GetAddrs(GetAddrsMessage::decode(cursor)?)),
-            PacketType::Addrs => Ok(Self::Addrs(AddrsMessage::decode(cursor)?)),
-            PacketType::Inv => Ok(Self::Inv(InvMessage::decode(cursor)?)),
-            PacketType::GetSlabs => Ok(Self::GetSlabs(GetSlabsMessage::decode(cursor)?)),
-            PacketType::Slab => Ok(Self::Slab(SlabMessage::decode(cursor)?)),
-            PacketType::Version => Ok(Self::Version(VersionMessage::decode(cursor)?)),
-            PacketType::Verack => Ok(Self::Verack(VerackMessage::decode(cursor)?)),
-        }
-    }
-
-    pub fn name(&self) -> &'static str {
-        match self {
-            Message::Ping(_) => "Ping",
-            Message::Pong(_) => "Pong",
-            Message::GetAddrs(_) => "GetAddrs",
-            Message::Addrs(_) => "Addrs",
-            Message::Inv(_) => "Inv",
-            Message::GetSlabs(_) => "GetSlabs",
-            Message::Slab(_) => "Slab",
-            Message::Version(_) => "Version",
-            Message::Verack(_) => "Verack",
-        }
-    }
-}
-
 // Packets are the base type read from the network
 // These are converted to messages and passed to event loop
 pub struct Packet {
-    pub command: PacketType,
+    pub command: String,
     pub payload: Vec<u8>,
 }
 
@@ -351,9 +170,13 @@ pub async fn read_packet<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Packet>
     }
 
     // The type of the message
-    let command = AsyncReadExt::read_u8(stream).await?;
+    let command_len = VarInt::decode_async(stream).await?.0 as usize;
+    let mut command = vec![0u8; command_len];
+    if command_len > 0 {
+        stream.read_exact(&mut command).await?;
+    }
+    let command = String::from_utf8(command)?;
     debug!(target: "net", "read command: {}", command);
-    let command = PacketType::try_from(command).map_err(|_| Error::MalformedPacket)?;
 
     let payload_len = VarInt::decode_async(stream).await?.0 as usize;
 
@@ -364,7 +187,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, payload })
+    Ok(Packet {
+        command: command,
+        payload,
+    })
 }
 
 pub async fn send_packet<W: AsyncWrite + Unpin>(stream: &mut W, packet: Packet) -> Result<()> {
@@ -372,8 +198,12 @@ pub async fn send_packet<W: AsyncWrite + Unpin>(stream: &mut W, packet: Packet)
     stream.write_all(&MAGIC_BYTES).await?;
     debug!(target: "net", "sent magic...");
 
-    AsyncWriteExt::write_u8(stream, packet.command as u8).await?;
-    debug!(target: "net", "sent command: {}", packet.command as u8);
+    VarInt(packet.command.len() as u64)
+        .encode_async(stream)
+        .await?;
+    assert!(!packet.command.is_empty());
+    stream.write_all(&packet.command.as_bytes()).await?;
+    debug!(target: "net", "sent command: {}", packet.command);
 
     assert_eq!(std::mem::size_of::<usize>(), std::mem::size_of::<u64>());
     VarInt(packet.payload.len() as u64)
@@ -387,82 +217,3 @@ pub async fn send_packet<W: AsyncWrite + Unpin>(stream: &mut W, packet: Packet)
 
     Ok(())
 }
-
-pub async fn receive_message<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Message> {
-    let packet = read_packet(stream).await?;
-    debug!(target: "net", "unpacking packet: {:?}", packet.command);
-    let message = Message::unpack(packet)?;
-    debug!(target: "net", "received Message::{}", message.name());
-    Ok(message)
-}
-
-pub async fn send_message<W: AsyncWrite + Unpin>(stream: &mut W, message: Message) -> Result<()> {
-    debug!(target: "net", "sending Message::{}", message.name());
-    let packet = message.pack()?;
-    send_packet(stream, packet).await
-}
-
-pub async fn sleep(seconds: u64) {
-    Timer::after(Duration::from_secs(seconds)).await;
-}
-
-// Used for ping pong loop timer
-pub struct InactivityTimer {
-    reset_sender: async_channel::Sender<()>,
-    timeout_receiver: async_channel::Receiver<()>,
-    task: smol::Task<()>,
-}
-
-impl InactivityTimer {
-    pub fn new(executor: Arc<Executor<'_>>) -> Self {
-        let (reset_sender, reset_receiver) = async_channel::bounded::<()>(1);
-        let (timeout_sender, timeout_receiver) = async_channel::bounded::<()>(1);
-
-        let task = executor.spawn(async {
-            match Self::_start(reset_receiver, timeout_sender).await {
-                Ok(()) => {}
-                Err(err) => error!("InactivityTimer fatal error {}", err),
-            }
-        });
-
-        Self {
-            reset_sender,
-            timeout_receiver,
-            task,
-        }
-    }
-
-    pub async fn stop(self) {
-        self.task.cancel().await;
-    }
-
-    // This loop basically waits for 10 secs. If it doesn't
-    // receive a signal that something happened then it will
-    // send a timeout signal. This will wakeup the main event loop
-    // and the connection will be dropped.
-    async fn _start(
-        reset_rx: async_channel::Receiver<()>,
-        timeout_sx: async_channel::Sender<()>,
-    ) -> Result<()> {
-        loop {
-            let is_awake = futures::select! {
-                _ = reset_rx.recv().fuse() => true,
-                _ = sleep(10).fuse() => false
-            };
-
-            if !is_awake {
-                warn!("InactivityTimer timeout");
-                timeout_sx.send(()).await?;
-            }
-        }
-    }
-
-    pub async fn reset(&self) -> Result<()> {
-        self.reset_sender.send(()).await?;
-        Ok(())
-    }
-
-    pub async fn wait_for_wakeup(&self) -> Result<()> {
-        Ok(self.timeout_receiver.recv().await?)
-    }
-}

src/net/mod.rs

@@ -1,11 +1,7 @@
-use smol::Async;
-use std::net::TcpStream;
-
 pub mod acceptor;
 pub mod channel;
 pub mod connector;
 pub mod error;
-#[macro_use]
 pub mod message_subscriber;
 pub mod hosts;
 pub mod messages;
@@ -15,12 +11,9 @@ pub mod sessions;
 pub mod settings;
 pub mod utility;
 
-pub type AsyncTcpStream = async_dup::Arc<Async<TcpStream>>;
-
 pub use acceptor::{Acceptor, AcceptorPtr};
 pub use channel::{Channel, ChannelPtr};
 pub use connector::Connector;
 pub use hosts::{Hosts, HostsPtr};
-pub use message_subscriber::{MessageSubscriber, MessageSubscription};
 pub use p2p::P2p;
 pub use settings::{Settings, SettingsPtr};

src/net/p2p.rs

@@ -1,7 +1,7 @@
 use async_executor::Executor;
 use async_std::sync::Mutex;
 use log::*;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 use std::sync::Arc;
 
@@ -10,13 +10,16 @@ use crate::net::sessions::{InboundSession, OutboundSession, SeedSession};
 use crate::net::{Channel, ChannelPtr, Hosts, HostsPtr, Settings, SettingsPtr};
 use crate::system::{Subscriber, SubscriberPtr, Subscription};
 
-pub type Pending<T> = Mutex<HashMap<SocketAddr, Arc<T>>>;
+pub type PendingChannels = Mutex<HashSet<SocketAddr>>;
+pub type ConnectedChannels<T> = Mutex<HashMap<SocketAddr, Arc<T>>>;
 
 pub type P2pPtr = Arc<P2p>;
 
 pub struct P2p {
-    pending_channels: Pending<Channel>,
-    // Used internally
+    pending: PendingChannels,
+    channels: ConnectedChannels<Channel>,
+    channel_subscriber: SubscriberPtr<NetResult<ChannelPtr>>,
+    // Used both internally and externally
     stop_subscriber: SubscriberPtr<NetError>,
     hosts: HostsPtr,
     settings: SettingsPtr,
@@ -26,9 +29,11 @@ impl P2p {
     pub fn new(settings: Settings) -> Arc<Self> {
         let settings = Arc::new(settings);
         Arc::new(Self {
-            pending_channels: Mutex::new(HashMap::new()),
+            pending: Mutex::new(HashSet::new()),
+            channels: Mutex::new(HashMap::new()),
+            channel_subscriber: Subscriber::new(),
             stop_subscriber: Subscriber::new(),
-            hosts: Hosts::new(settings.clone()),
+            hosts: Hosts::new(),
             settings,
         })
     }
@@ -50,6 +55,8 @@ impl P2p {
     /// Synchronize the blockchain and then begin long running sessions,
     /// call after start() is invoked.
     pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
+        debug!(target: "net", "P2p::run() [BEGIN]");
+
         let inbound = InboundSession::new(Arc::downgrade(&self));
         inbound.clone().start(executor.clone())?;
 
@@ -64,24 +71,34 @@ impl P2p {
         inbound.stop().await;
         outbound.stop().await;
 
+        debug!(target: "net", "P2p::run() [BEGIN]");
         Ok(())
     }
 
-    pub async fn store(self: Arc<Self>, channel: ChannelPtr) {
-        self.pending_channels
+    pub async fn store(&self, channel: ChannelPtr) {
+        self.channels
             .lock()
             .await
-            .insert(channel.address(), channel);
+            .insert(channel.address(), channel.clone());
+        self.channel_subscriber.notify(Ok(channel)).await;
     }
-    pub async fn remove(self: Arc<Self>, channel: ChannelPtr) {
-        self.pending_channels
-            .lock()
-            .await
-            .remove(&channel.address());
+    pub async fn remove(&self, channel: ChannelPtr) {
+        self.channels.lock().await.remove(&channel.address());
+    }
+
+    pub async fn exists(&self, addr: &SocketAddr) -> bool {
+        self.channels.lock().await.contains_key(addr)
+    }
+
+    pub async fn add_pending(&self, addr: SocketAddr) -> bool {
+        self.pending.lock().await.insert(addr)
+    }
+    pub async fn remove_pending(&self, addr: &SocketAddr) {
+        self.pending.lock().await.remove(addr);
     }
 
     pub async fn connections_count(&self) -> usize {
-        self.pending_channels.lock().await.len()
+        self.channels.lock().await.len()
     }
 
     pub fn settings(&self) -> SettingsPtr {
@@ -92,7 +109,11 @@ impl P2p {
         self.hosts.clone()
     }
 
-    async fn subscribe_stop(&self) -> Subscription<NetError> {
+    pub async fn subscribe_channel(&self) -> Subscription<NetResult<ChannelPtr>> {
+        self.channel_subscriber.clone().subscribe().await
+    }
+
+    pub async fn subscribe_stop(&self) -> Subscription<NetError> {
         self.stop_subscriber.clone().subscribe().await
     }
 }

src/net/protocols/protocol_address.rs

@@ -6,38 +6,38 @@ use crate::net::error::NetResult;
 use crate::net::message_subscriber::MessageSubscription;
 use crate::net::messages;
 use crate::net::protocols::{ProtocolJobsManager, ProtocolJobsManagerPtr};
-use crate::net::{ChannelPtr, HostsPtr, SettingsPtr};
+use crate::net::{ChannelPtr, HostsPtr};
 
 pub struct ProtocolAddress {
     channel: ChannelPtr,
 
-    addrs_sub: MessageSubscription,
-    get_addrs_sub: MessageSubscription,
+    addrs_sub: MessageSubscription<messages::AddrsMessage>,
+    get_addrs_sub: MessageSubscription<messages::GetAddrsMessage>,
 
     hosts: HostsPtr,
-    settings: SettingsPtr,
 
     jobsman: ProtocolJobsManagerPtr,
 }
 
 impl ProtocolAddress {
-    pub async fn new(channel: ChannelPtr, hosts: HostsPtr, settings: SettingsPtr) -> Arc<Self> {
+    pub async fn new(channel: ChannelPtr, hosts: HostsPtr) -> Arc<Self> {
         let addrs_sub = channel
             .clone()
-            .subscribe_msg(messages::PacketType::Addrs)
-            .await;
+            .subscribe_msg::<messages::AddrsMessage>()
+            .await
+            .expect("Missing addrs dispatcher!");
 
         let get_addrs_sub = channel
             .clone()
-            .subscribe_msg(messages::PacketType::GetAddrs)
-            .await;
+            .subscribe_msg::<messages::GetAddrsMessage>()
+            .await
+            .expect("Missing getaddrs dispatcher!");
 
         Arc::new(Self {
             channel: channel.clone(),
             addrs_sub,
             get_addrs_sub,
             hosts,
-            settings,
             jobsman: ProtocolJobsManager::new("ProtocolAddress", channel),
         })
     }
@@ -55,7 +55,7 @@ impl ProtocolAddress {
             .await;
 
         // Send get_address message
-        let get_addrs = messages::Message::GetAddrs(messages::GetAddrsMessage {});
+        let get_addrs = messages::GetAddrsMessage {};
         let _ = self.channel.clone().send(get_addrs).await;
         debug!(target: "net", "ProtocolAddress::start() [END]");
     }
@@ -63,9 +63,16 @@ impl ProtocolAddress {
     async fn handle_receive_addrs(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net", "ProtocolAddress::handle_receive_addrs() [START]");
         loop {
-            let addrs_msg = receive_message!(self.addrs_sub, messages::Message::Addrs);
+            let addrs_msg = self.addrs_sub.receive().await?;
 
-            debug!(target: "net", "ProtocolAddress::handle_receive_addrs() storing address in hosts");
+            debug!(
+                target: "net",
+                "ProtocolAddress::handle_receive_addrs() received {} addrs",
+                addrs_msg.addrs.len()
+            );
+            for (i, addr) in addrs_msg.addrs.iter().enumerate() {
+                debug!("  addr[{}]: {}", i, addr);
+            }
             self.hosts.store(addrs_msg.addrs.clone()).await;
         }
     }
@@ -73,15 +80,18 @@ impl ProtocolAddress {
     async fn handle_receive_get_addrs(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() [START]");
         loop {
-            let _get_addrs = receive_message!(self.get_addrs_sub, messages::Message::GetAddrs);
+            let _get_addrs = self.get_addrs_sub.receive().await?;
 
             debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() received GetAddrs message");
 
-            let addrs = messages::Message::Addrs(messages::AddrsMessage {
-                addrs: self.hosts.load_all().await,
-            });
-            debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() sending Addrs message");
-            self.channel.clone().send(addrs).await?;
+            let addrs = self.hosts.load_all().await;
+            debug!(
+                target: "net",
+                "ProtocolAddress::handle_receive_get_addrs() sending {} addrs",
+                addrs.len()
+            );
+            let addrs_msg = messages::AddrsMessage { addrs };
+            self.channel.clone().send(addrs_msg).await?;
         }
     }
 }

src/net/protocols/protocol_jobs_manager.rs

@@ -29,6 +29,7 @@ impl ProtocolJobsManager {
         executor.spawn(self.handle_stop()).detach()
     }
 
+    /// Spawns a new task adding it to the internal queue
     pub async fn spawn<'a, F>(&self, future: F, executor: ExecutorPtr<'a>)
     where
         F: Future<Output = NetResult<()>> + Send + 'a,
@@ -36,6 +37,7 @@ impl ProtocolJobsManager {
         self.tasks.lock().await.push(executor.spawn(future))
     }
 
+    /// This is run in start(). When the channel closes, we also stop all the tasks
     async fn handle_stop(self: Arc<Self>) {
         let stop_sub = self.channel.clone().subscribe_stop().await;
 
@@ -52,8 +54,10 @@ impl ProtocolJobsManager {
             self.name,
             self.channel.address()
         );
+        // Take all the tasks from our internal queue...
         let tasks = std::mem::take(&mut *self.tasks.lock().await);
         for task in tasks {
+            // ... and cancel them
             let _ = task.cancel().await;
         }
     }

src/net/protocols/protocol_ping.rs

@@ -2,6 +2,7 @@ use log::*;
 use rand::Rng;
 use smol::Executor;
 use std::sync::Arc;
+use std::time::Instant;
 
 use crate::net::error::{NetError, NetResult};
 use crate::net::messages;
@@ -44,8 +45,9 @@ impl ProtocolPing {
         let pong_sub = self
             .channel
             .clone()
-            .subscribe_msg(messages::PacketType::Pong)
-            .await;
+            .subscribe_msg::<messages::PongMessage>()
+            .await
+            .expect("Missing pong dispatcher!");
 
         loop {
             // Wait channel_heartbeat amount of time
@@ -55,18 +57,21 @@ impl ProtocolPing {
             let nonce = Self::random_nonce();
 
             // Send ping message
-            let ping = messages::Message::Ping(messages::PingMessage { nonce });
+            let ping = messages::PingMessage { nonce };
             self.channel.clone().send(ping).await?;
             debug!(target: "net", "ProtocolPing::run_ping_pong() send Ping message");
+            // Start the timer for ping timer
+            let start = Instant::now();
 
             // Wait for pong, check nonce matches
-            let pong_msg = receive_message!(pong_sub, messages::Message::Pong);
+            let pong_msg = pong_sub.receive().await?;
             if pong_msg.nonce != nonce {
                 error!("Wrong nonce for ping reply. Disconnecting from channel.");
                 self.channel.stop().await;
                 return Err(NetError::ChannelStopped);
             }
-            debug!(target: "net", "ProtocolPing::run_ping_pong() received Pong message");
+            let duration = start.elapsed().as_millis();
+            debug!(target: "net", "Received Pong message {}ms from [{:?}]", duration, self.channel.address());
         }
     }
 
@@ -75,16 +80,17 @@ impl ProtocolPing {
         let ping_sub = self
             .channel
             .clone()
-            .subscribe_msg(messages::PacketType::Ping)
-            .await;
+            .subscribe_msg::<messages::PingMessage>()
+            .await
+            .expect("Missing ping dispatcher!");
 
         loop {
             // Wait for ping, reply with pong that has a matching nonce
-            let ping = receive_message!(ping_sub, messages::Message::Ping);
+            let ping = ping_sub.receive().await?;
             debug!(target: "net", "ProtocolPing::reply_to_ping() received Ping message");
 
             // Send ping message
-            let pong = messages::Message::Pong(messages::PongMessage { nonce: ping.nonce });
+            let pong = messages::PongMessage { nonce: ping.nonce };
             self.channel.clone().send(pong).await?;
             debug!(target: "net", "ProtocolPing::reply_to_ping() sent Pong reply");
         }

src/net/protocols/protocol_seed.rs

@@ -26,28 +26,31 @@ impl ProtocolSeed {
         let addr_sub = self
             .channel
             .clone()
-            .subscribe_msg(messages::PacketType::Addrs)
-            .await;
+            .subscribe_msg::<messages::AddrsMessage>()
+            .await
+            .expect("Missing addrs dispatcher!");
 
         // Send own address to the seed server
-        self.send_own_address().await?;
+        self.send_self_address().await?;
 
         // Send get address message
-        let get_addr = messages::Message::GetAddrs(messages::GetAddrsMessage {});
+        let get_addr = messages::GetAddrsMessage {};
         self.channel.clone().send(get_addr).await?;
 
         // Receive addresses
-        let addrs_msg = receive_message!(addr_sub, messages::Message::Addrs);
+        let addrs_msg = addr_sub.receive().await?;
+        debug!(target: "net", "ProtocolSeed::start() received {} addrs", addrs_msg.addrs.len());
         self.hosts.store(addrs_msg.addrs.clone()).await;
 
         debug!(target: "net", "ProtocolSeed::start() [END]");
         Ok(())
     }
 
-    pub async fn send_own_address(&self) -> NetResult<()> {
+    pub async fn send_self_address(&self) -> NetResult<()> {
         match self.settings.external_addr {
             Some(addr) => {
-                let addr = messages::Message::Addrs(messages::AddrsMessage { addrs: vec![addr] });
+                debug!(target: "net", "ProtocolSeed::send_own_address() addr={}", addr);
+                let addr = messages::AddrsMessage { addrs: vec![addr] };
                 self.channel.clone().send(addr).await?;
             }
             None => {

src/net/protocols/protocol_version.rs

@@ -11,8 +11,8 @@ use crate::net::{ChannelPtr, SettingsPtr};
 
 pub struct ProtocolVersion {
     channel: ChannelPtr,
-    version_sub: MessageSubscription,
-    verack_sub: MessageSubscription,
+    version_sub: MessageSubscription<messages::VersionMessage>,
+    verack_sub: MessageSubscription<messages::VerackMessage>,
     settings: SettingsPtr,
 }
 
@@ -20,13 +20,15 @@ impl ProtocolVersion {
     pub async fn new(channel: ChannelPtr, settings: SettingsPtr) -> Arc<Self> {
         let version_sub = channel
             .clone()
-            .subscribe_msg(messages::PacketType::Version)
-            .await;
+            .subscribe_msg::<messages::VersionMessage>()
+            .await
+            .expect("Missing version dispatcher!");
 
         let verack_sub = channel
             .clone()
-            .subscribe_msg(messages::PacketType::Verack)
-            .await;
+            .subscribe_msg::<messages::VerackMessage>()
+            .await
+            .expect("Missing verack dispatcher!");
 
         Arc::new(Self {
             channel,
@@ -63,7 +65,7 @@ impl ProtocolVersion {
 
     async fn send_version(self: Arc<Self>) -> NetResult<()> {
         debug!(target: "net", "ProtocolVersion::send_version() [START]");
-        let version = messages::Message::Version(messages::VersionMessage {});
+        let version = messages::VersionMessage {};
         self.channel.clone().send(version).await?;
 
         // Wait for version acknowledgement
@@ -80,7 +82,7 @@ impl ProtocolVersion {
         // Check the message is OK
 
         // Send version acknowledgement
-        let verack = messages::Message::Verack(messages::VerackMessage {});
+        let verack = messages::VerackMessage {};
         self.channel.clone().send(verack).await?;
 
         debug!(target: "net", "ProtocolVersion::recv_version() [END]");

src/net/sessions/inbound_session.rs

@@ -18,12 +18,7 @@ pub struct InboundSession {
 
 impl InboundSession {
     pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
-        let settings = {
-            let p2p = p2p.upgrade().unwrap();
-            p2p.settings()
-        };
-
-        let acceptor = Acceptor::new(settings);
+        let acceptor = Acceptor::new();
 
         Arc::new(Self {
             p2p,
@@ -73,10 +68,11 @@ impl InboundSession {
         result
     }
 
+    /// 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 {
-            let channel = (*channel_sub.receive().await).clone()?;
+            let channel = channel_sub.receive().await?;
             // Spawn a detached task to process the channel
             // This will just perform the channel setup then exit.
             executor
@@ -108,7 +104,7 @@ impl InboundSession {
         let hosts = self.p2p().hosts().clone();
 
         let protocol_ping = ProtocolPing::new(channel.clone(), settings.clone());
-        let protocol_addr = ProtocolAddress::new(channel, hosts, settings).await;
+        let protocol_addr = ProtocolAddress::new(channel, hosts).await;
 
         protocol_ping.start(executor.clone()).await;
         protocol_addr.start(executor).await;

src/net/sessions/outbound_session.rs

@@ -25,6 +25,7 @@ impl OutboundSession {
 
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         let slots_count = self.p2p().settings().outbound_connections;
+        info!("Starting {} outbound connection slots.", slots_count);
         let mut connect_slots = self.connect_slots.lock().await;
 
         for i in 0..slots_count {
@@ -61,13 +62,13 @@ impl OutboundSession {
 
         loop {
             let addr = self.load_address(slot_number).await?;
-            info!("Connecting to outbound [{}]", addr);
+            info!("#{} connecting to outbound [{}]", slot_number, addr);
 
             match connector.connect(addr).await {
                 Ok(channel) => {
                     // Blacklist goes here
 
-                    info!("Connected outbound [{}]", addr);
+                    info!("#{} connected to outbound [{}]", slot_number, addr);
 
                     let stop_sub = channel.subscribe_stop().await;
 
@@ -75,6 +76,11 @@ impl OutboundSession {
                         .register_channel(channel.clone(), executor.clone())
                         .await?;
 
+                    // Channel is now connected but not yet setup
+
+                    // Remove pending lock since register_channel will add the channel to p2p
+                    self.p2p().remove_pending(&addr).await;
+
                     self.clone()
                         .attach_protocols(channel, executor.clone())
                         .await?;
@@ -89,18 +95,49 @@ impl OutboundSession {
         }
     }
 
+    /// Load a valid address that we can connect to.
+    /// Valid means we aren't connecting (pending state) or connected (open channel)
+    /// in another slot, and it isn't our own inbound address.
+    /// Retry otherwise.
     async fn load_address(&self, slot_number: u32) -> NetResult<SocketAddr> {
-        let hosts = self.p2p().hosts();
+        let p2p = self.p2p();
+        let hosts = p2p.hosts();
+        let inbound_addr = p2p.settings().inbound;
+
+        loop {
+            let addr = hosts.load_single().await;
 
-        match hosts.load_single().await {
-            Some(addr) => Ok(addr),
-            None => {
+            if addr.is_none() {
                 error!(
                     "Hosts address pool is empty. Closing connect slot #{}",
                     slot_number
                 );
-                Err(NetError::ServiceStopped)
+                return Err(NetError::ServiceStopped);
+            }
+            let addr = addr.unwrap();
+
+            if Self::addr_is_inbound(&addr, &inbound_addr) {
+                continue;
+            }
+
+            if p2p.exists(&addr).await {
+                continue;
+            }
+
+            // Obtain a lock on this address to prevent duplicate connections
+            if !p2p.add_pending(addr).await {
+                continue;
             }
+
+            return Ok(addr);
+        }
+    }
+
+    fn addr_is_inbound(addr: &SocketAddr, inbound_addr: &Option<SocketAddr>) -> bool {
+        match inbound_addr {
+            Some(inbound_addr) => inbound_addr == addr,
+            // No inbound listening address configured
+            None => false,
         }
     }
 
@@ -113,7 +150,7 @@ impl OutboundSession {
         let hosts = self.p2p().hosts().clone();
 
         let protocol_ping = ProtocolPing::new(channel.clone(), settings.clone());
-        let protocol_addr = ProtocolAddress::new(channel, hosts, settings).await;
+        let protocol_addr = ProtocolAddress::new(channel, hosts).await;
 
         protocol_ping.start(executor.clone()).await;
         protocol_addr.start(executor).await;

src/net/sessions/seed_session.rs

@@ -1,4 +1,5 @@
 use async_executor::Executor;
+use futures::FutureExt;
 use log::*;
 use std::net::SocketAddr;
 use std::sync::{Arc, Weak};
@@ -6,6 +7,7 @@ use std::sync::{Arc, Weak};
 use crate::net::error::{NetError, NetResult};
 use crate::net::protocols::{ProtocolPing, ProtocolSeed};
 use crate::net::sessions::Session;
+use crate::net::utility::sleep;
 use crate::net::{ChannelPtr, Connector, HostsPtr, P2p, SettingsPtr};
 
 pub struct SeedSession {
@@ -19,40 +21,46 @@ impl SeedSession {
 
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         debug!(target: "net", "SeedSession::start() [START]");
-        let settings = {
-            let p2p = self.p2p.upgrade().unwrap();
-            p2p.settings()
-        };
+        let settings = self.p2p().settings();
 
-        if settings.skip_seed_sync {
-            info!("Configured to skip seed synchronization process.");
+        if settings.seeds.is_empty() {
+            warn!("Skipping seed sync process since no seeds are configured.");
             return Ok(());
         }
 
         // if cached addresses then quit
 
-        // if seeds empty then seeding required but empty
-        if settings.seeds.is_empty() {
-            error!("Seeding is required but no seeds are configured.");
-            return Err(NetError::OperationFailed);
-        }
-
         let mut tasks = Vec::new();
 
         for (i, seed) in settings.seeds.iter().enumerate() {
             tasks.push(executor.spawn(self.clone().start_seed(i, seed.clone(), executor.clone())));
         }
 
-        for (i, task) in tasks.into_iter().enumerate() {
-            // Ignore errors
-            match task.await {
-                Ok(()) => info!("Successfully queried seed #{}", i),
-                Err(err) => warn!("Seed query #{} failed for reason: {}", i, err),
+        // This line loops through all the tasks and waits for them to finish.
+        // But if the seed_query_timeout_seconds times out before they are finished,
+        // then it will simply quit and the tasks will get dropped.
+        futures::select! {
+            _ = async move {
+                for (i, task) in tasks.into_iter().enumerate() {
+                    // Ignore errors
+                    match task.await {
+                        Ok(()) => info!("Successfully queried seed #{}", i),
+                        Err(err) => warn!("Seed query #{} failed for reason: {}", i, err),
+                    }
+                }
+            }.fuse() => {
+            }
+            _ = sleep(settings.seed_query_timeout_seconds).fuse() => {
+                error!("Querying seeds timed out");
+                return Err(NetError::OperationFailed);
             }
         }
 
         // Seed process complete
-        // TODO: check increase count of address
+        if self.p2p().hosts().is_empty().await {
+            error!("Hosts pool still empty after seeding");
+            return Err(NetError::OperationFailed);
+        }
 
         debug!(target: "net", "SeedSession::start() [END]");
         Ok(())

src/net/sessions/session.rs

@@ -57,7 +57,7 @@ pub trait Session: Sync {
         // Channel is now initialized
 
         // Add channel to p2p
-        self.p2p().clone().store(channel.clone()).await;
+        self.p2p().store(channel.clone()).await;
 
         // Subscribe to stop, so can remove from p2p
         executor

src/net/settings.rs

@@ -8,6 +8,7 @@ pub struct Settings {
     pub inbound: Option<SocketAddr>,
     pub outbound_connections: u32,
 
+    pub seed_query_timeout_seconds: u32,
     pub connect_timeout_seconds: u32,
     pub channel_handshake_seconds: u32,
     pub channel_heartbeat_seconds: u32,
@@ -15,5 +16,4 @@ pub struct Settings {
     pub external_addr: Option<SocketAddr>,
     pub peers: Vec<SocketAddr>,
     pub seeds: Vec<SocketAddr>,
-    pub skip_seed_sync: bool,
 }

src/serial.rs

@@ -9,185 +9,185 @@ use crate::error::{Error, Result};
 
 /// Encode an object into a vector
 pub fn serialize<T: Encodable + ?Sized>(data: &T) -> Vec<u8> {
-let mut encoder = Vec::new();
-let len = data.encode(&mut encoder).unwrap();
-assert_eq!(len, encoder.len());
-encoder
+    let mut encoder = Vec::new();
+    let len = data.encode(&mut encoder).unwrap();
+    assert_eq!(len, encoder.len());
+    encoder
 }
 
 /// Encode an object into a hex-encoded string
 pub fn serialize_hex<T: Encodable + ?Sized>(data: &T) -> String {
-hex::encode(serialize(data))
+    hex::encode(serialize(data))
 }
 
 /// Deserialize an object from a vector, will error if said deserialization
 /// doesn't consume the entire vector.
 pub fn deserialize<T: Decodable>(data: &[u8]) -> Result<T> {
-let (rv, consumed) = deserialize_partial(data)?;
+    let (rv, consumed) = deserialize_partial(data)?;
 
-// Fail if data are not consumed entirely.
-if consumed == data.len() {
-Ok(rv)
-} else {
-Err(Error::ParseFailed(
-"data not consumed entirely when explicitly deserializing",
-))
-}
+    // Fail if data are not consumed entirely.
+    if consumed == data.len() {
+        Ok(rv)
+    } else {
+        Err(Error::ParseFailed(
+            "data not consumed entirely when explicitly deserializing",
+        ))
+    }
 }
 
 /// Deserialize an object from a vector, but will not report an error if said deserialization
 /// doesn't consume the entire vector.
 pub fn deserialize_partial<T: Decodable>(data: &[u8]) -> Result<(T, usize)> {
-let mut decoder = Cursor::new(data);
-let rv = Decodable::decode(&mut decoder)?;
-let consumed = decoder.position() as usize;
+    let mut decoder = Cursor::new(data);
+    let rv = Decodable::decode(&mut decoder)?;
+    let consumed = decoder.position() as usize;
 
-Ok((rv, consumed))
+    Ok((rv, consumed))
 }
 
 /// Extensions of `Write` to encode data as per Bitcoin consensus
 pub trait WriteExt {
-/// Output a 64-bit uint
-fn write_u64(&mut self, v: u64) -> Result<()>;
-/// Output a 32-bit uint
-fn write_u32(&mut self, v: u32) -> Result<()>;
-/// Output a 16-bit uint
-fn write_u16(&mut self, v: u16) -> Result<()>;
-/// Output a 8-bit uint
-fn write_u8(&mut self, v: u8) -> Result<()>;
-
-/// Output a 64-bit int
-fn write_i64(&mut self, v: i64) -> Result<()>;
-/// Output a 32-bit int
-fn write_i32(&mut self, v: i32) -> Result<()>;
-/// Output a 16-bit int
-fn write_i16(&mut self, v: i16) -> Result<()>;
-/// Output a 8-bit int
-fn write_i8(&mut self, v: i8) -> Result<()>;
-
-/// Output a boolean
-fn write_bool(&mut self, v: bool) -> Result<()>;
-
-/// Output a byte slice
-fn write_slice(&mut self, v: &[u8]) -> Result<()>;
+    /// Output a 64-bit uint
+    fn write_u64(&mut self, v: u64) -> Result<()>;
+    /// Output a 32-bit uint
+    fn write_u32(&mut self, v: u32) -> Result<()>;
+    /// Output a 16-bit uint
+    fn write_u16(&mut self, v: u16) -> Result<()>;
+    /// Output a 8-bit uint
+    fn write_u8(&mut self, v: u8) -> Result<()>;
+
+    /// Output a 64-bit int
+    fn write_i64(&mut self, v: i64) -> Result<()>;
+    /// Output a 32-bit int
+    fn write_i32(&mut self, v: i32) -> Result<()>;
+    /// Output a 16-bit int
+    fn write_i16(&mut self, v: i16) -> Result<()>;
+    /// Output a 8-bit int
+    fn write_i8(&mut self, v: i8) -> Result<()>;
+
+    /// Output a boolean
+    fn write_bool(&mut self, v: bool) -> Result<()>;
+
+    /// Output a byte slice
+    fn write_slice(&mut self, v: &[u8]) -> Result<()>;
 }
 
 /// Extensions of `Read` to decode data as per Bitcoin consensus
 pub trait ReadExt {
-/// Read a 64-bit uint
-fn read_u64(&mut self) -> Result<u64>;
-/// Read a 32-bit uint
-fn read_u32(&mut self) -> Result<u32>;
-/// Read a 16-bit uint
-fn read_u16(&mut self) -> Result<u16>;
-/// Read a 8-bit uint
-fn read_u8(&mut self) -> Result<u8>;
-
-/// Read a 64-bit int
-fn read_i64(&mut self) -> Result<i64>;
-/// Read a 32-bit int
-fn read_i32(&mut self) -> Result<i32>;
-/// Read a 16-bit int
-fn read_i16(&mut self) -> Result<i16>;
-/// Read a 8-bit int
-fn read_i8(&mut self) -> Result<i8>;
-
-/// Read a boolean
-fn read_bool(&mut self) -> Result<bool>;
-
-/// Read a byte slice
-fn read_slice(&mut self, slice: &mut [u8]) -> Result<()>;
+    /// Read a 64-bit uint
+    fn read_u64(&mut self) -> Result<u64>;
+    /// Read a 32-bit uint
+    fn read_u32(&mut self) -> Result<u32>;
+    /// Read a 16-bit uint
+    fn read_u16(&mut self) -> Result<u16>;
+    /// Read a 8-bit uint
+    fn read_u8(&mut self) -> Result<u8>;
+
+    /// Read a 64-bit int
+    fn read_i64(&mut self) -> Result<i64>;
+    /// Read a 32-bit int
+    fn read_i32(&mut self) -> Result<i32>;
+    /// Read a 16-bit int
+    fn read_i16(&mut self) -> Result<i16>;
+    /// Read a 8-bit int
+    fn read_i8(&mut self) -> Result<i8>;
+
+    /// Read a boolean
+    fn read_bool(&mut self) -> Result<bool>;
+
+    /// Read a byte slice
+    fn read_slice(&mut self, slice: &mut [u8]) -> Result<()>;
 }
 
 macro_rules! encoder_fn {
-($name:ident, $val_type:ty, $writefn:ident) => {
-#[inline]
-fn $name(&mut self, v: $val_type) -> Result<()> {
-self.write_all(&endian::$writefn(v)).map_err(Error::Io)
-}
-};
+    ($name:ident, $val_type:ty, $writefn:ident) => {
+        #[inline]
+        fn $name(&mut self, v: $val_type) -> Result<()> {
+            self.write_all(&endian::$writefn(v)).map_err(Error::Io)
+        }
+    };
 }
 
 macro_rules! decoder_fn {
-($name:ident, $val_type:ty, $readfn:ident, $byte_len: expr) => {
-#[inline]
-fn $name(&mut self) -> Result<$val_type> {
-assert_eq!(::std::mem::size_of::<$val_type>(), $byte_len); // size_of isn't a constfn in 1.22
-let mut val = [0; $byte_len];
-self.read_exact(&mut val[..]).map_err(Error::Io)?;
-Ok(endian::$readfn(&val))
-}
-};
+    ($name:ident, $val_type:ty, $readfn:ident, $byte_len: expr) => {
+        #[inline]
+        fn $name(&mut self) -> Result<$val_type> {
+            assert_eq!(::std::mem::size_of::<$val_type>(), $byte_len); // size_of isn't a constfn in 1.22
+            let mut val = [0; $byte_len];
+            self.read_exact(&mut val[..]).map_err(Error::Io)?;
+            Ok(endian::$readfn(&val))
+        }
+    };
 }
 
 impl<W: Write> WriteExt for W {
-encoder_fn!(write_u64, u64, u64_to_array_le);
-encoder_fn!(write_u32, u32, u32_to_array_le);
-encoder_fn!(write_u16, u16, u16_to_array_le);
-encoder_fn!(write_i64, i64, i64_to_array_le);
-encoder_fn!(write_i32, i32, i32_to_array_le);
-encoder_fn!(write_i16, i16, i16_to_array_le);
-
-#[inline]
-fn write_i8(&mut self, v: i8) -> Result<()> {
-self.write_all(&[v as u8]).map_err(Error::Io)
-}
-#[inline]
-fn write_u8(&mut self, v: u8) -> Result<()> {
-self.write_all(&[v]).map_err(Error::Io)
-}
-#[inline]
-fn write_bool(&mut self, v: bool) -> Result<()> {
-self.write_all(&[v as u8]).map_err(Error::Io)
-}
-#[inline]
-fn write_slice(&mut self, v: &[u8]) -> Result<()> {
-self.write_all(v).map_err(Error::Io)
-}
+    encoder_fn!(write_u64, u64, u64_to_array_le);
+    encoder_fn!(write_u32, u32, u32_to_array_le);
+    encoder_fn!(write_u16, u16, u16_to_array_le);
+    encoder_fn!(write_i64, i64, i64_to_array_le);
+    encoder_fn!(write_i32, i32, i32_to_array_le);
+    encoder_fn!(write_i16, i16, i16_to_array_le);
+
+    #[inline]
+    fn write_i8(&mut self, v: i8) -> Result<()> {
+        self.write_all(&[v as u8]).map_err(Error::Io)
+    }
+    #[inline]
+    fn write_u8(&mut self, v: u8) -> Result<()> {
+        self.write_all(&[v]).map_err(Error::Io)
+    }
+    #[inline]
+    fn write_bool(&mut self, v: bool) -> Result<()> {
+        self.write_all(&[v as u8]).map_err(Error::Io)
+    }
+    #[inline]
+    fn write_slice(&mut self, v: &[u8]) -> Result<()> {
+        self.write_all(v).map_err(Error::Io)
+    }
 }
 
 impl<R: Read> ReadExt for R {
-decoder_fn!(read_u64, u64, slice_to_u64_le, 8);
-decoder_fn!(read_u32, u32, slice_to_u32_le, 4);
-decoder_fn!(read_u16, u16, slice_to_u16_le, 2);
-decoder_fn!(read_i64, i64, slice_to_i64_le, 8);
-decoder_fn!(read_i32, i32, slice_to_i32_le, 4);
-decoder_fn!(read_i16, i16, slice_to_i16_le, 2);
-
-#[inline]
-fn read_u8(&mut self) -> Result<u8> {
-let mut slice = [0u8; 1];
-self.read_exact(&mut slice)?;
-Ok(slice[0])
-}
-#[inline]
-fn read_i8(&mut self) -> Result<i8> {
-let mut slice = [0u8; 1];
-self.read_exact(&mut slice)?;
-Ok(slice[0] as i8)
-}
-#[inline]
-fn read_bool(&mut self) -> Result<bool> {
-ReadExt::read_i8(self).map(|bit| bit != 0)
-}
-#[inline]
-fn read_slice(&mut self, slice: &mut [u8]) -> Result<()> {
-self.read_exact(slice).map_err(Error::Io)
-}
+    decoder_fn!(read_u64, u64, slice_to_u64_le, 8);
+    decoder_fn!(read_u32, u32, slice_to_u32_le, 4);
+    decoder_fn!(read_u16, u16, slice_to_u16_le, 2);
+    decoder_fn!(read_i64, i64, slice_to_i64_le, 8);
+    decoder_fn!(read_i32, i32, slice_to_i32_le, 4);
+    decoder_fn!(read_i16, i16, slice_to_i16_le, 2);
+
+    #[inline]
+    fn read_u8(&mut self) -> Result<u8> {
+        let mut slice = [0u8; 1];
+        self.read_exact(&mut slice)?;
+        Ok(slice[0])
+    }
+    #[inline]
+    fn read_i8(&mut self) -> Result<i8> {
+        let mut slice = [0u8; 1];
+        self.read_exact(&mut slice)?;
+        Ok(slice[0] as i8)
+    }
+    #[inline]
+    fn read_bool(&mut self) -> Result<bool> {
+        ReadExt::read_i8(self).map(|bit| bit != 0)
+    }
+    #[inline]
+    fn read_slice(&mut self, slice: &mut [u8]) -> Result<()> {
+        self.read_exact(slice).map_err(Error::Io)
+    }
 }
 
 /// Data which can be encoded in a consensus-consistent way
 pub trait Encodable {
-/// Encode an object with a well-defined format, should only ever error if
-/// the underlying `Write` errors. Returns the number of bytes written on
-/// success
-fn encode<W: io::Write>(&self, e: W) -> Result<usize>;
+    /// Encode an object with a well-defined format, should only ever error if
+    /// the underlying `Write` errors. Returns the number of bytes written on
+    /// success
+    fn encode<W: io::Write>(&self, e: W) -> Result<usize>;
 }
 
 /// Data which can be encoded in a consensus-consistent way
 pub trait Decodable: Sized {
-/// Decode an object with a well-defined format
-fn decode<D: io::Read>(d: D) -> Result<Self>;
+    /// Decode an object with a well-defined format
+    fn decode<D: io::Read>(d: D) -> Result<Self>;
 }
 
 #[derive(PartialEq, Eq, PartialOrd, Ord, Clone, Debug)]
@@ -195,21 +195,21 @@ pub struct VarInt(pub u64);
 
 // Primitive types
 macro_rules! impl_int_encodable {
-($ty:ident, $meth_dec:ident, $meth_enc:ident) => {
-impl Decodable for $ty {
-#[inline]
-fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-ReadExt::$meth_dec(&mut d).map($ty::from_le)
-}
-}
-impl Encodable for $ty {
-#[inline]
-fn encode<S: WriteExt>(&self, mut s: S) -> Result<usize> {
-s.$meth_enc(self.to_le())?;
-Ok(mem::size_of::<$ty>())
-}
-}
-};
+    ($ty:ident, $meth_dec:ident, $meth_enc:ident) => {
+        impl Decodable for $ty {
+            #[inline]
+            fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+                ReadExt::$meth_dec(&mut d).map($ty::from_le)
+            }
+        }
+        impl Encodable for $ty {
+            #[inline]
+            fn encode<S: WriteExt>(&self, mut s: S) -> Result<usize> {
+                s.$meth_enc(self.to_le())?;
+                Ok(mem::size_of::<$ty>())
+            }
+        }
+    };
 }
 
 impl_int_encodable!(u8, read_u8, write_u8);
@@ -222,156 +222,156 @@ impl_int_encodable!(i32, read_i32, write_i32);
 impl_int_encodable!(i64, read_i64, write_i64);
 
 impl VarInt {
-/// Gets the length of this VarInt when encoded.
-/// Returns 1 for 0...0xFC, 3 for 0xFD...(2^16-1), 5 for 0x10000...(2^32-1),
-/// and 9 otherwise.
-#[inline]
-pub fn len(&self) -> usize {
-match self.0 {
-0..=0xFC => 1,
-0xFD..=0xFFFF => 3,
-0x10000..=0xFFFFFFFF => 5,
-_ => 9,
-}
-}
+    /// Gets the length of this VarInt when encoded.
+    /// Returns 1 for 0...0xFC, 3 for 0xFD...(2^16-1), 5 for 0x10000...(2^32-1),
+    /// and 9 otherwise.
+    #[inline]
+    pub fn len(&self) -> usize {
+        match self.0 {
+            0..=0xFC => 1,
+            0xFD..=0xFFFF => 3,
+            0x10000..=0xFFFFFFFF => 5,
+            _ => 9,
+        }
+    }
 }
 
 impl Encodable for VarInt {
-#[inline]
-fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-match self.0 {
-0..=0xFC => {
-(self.0 as u8).encode(s)?;
-Ok(1)
-}
-0xFD..=0xFFFF => {
-s.write_u8(0xFD)?;
-(self.0 as u16).encode(s)?;
-Ok(3)
-}
-0x10000..=0xFFFFFFFF => {
-s.write_u8(0xFE)?;
-(self.0 as u32).encode(s)?;
-Ok(5)
-}
-_ => {
-s.write_u8(0xFF)?;
-(self.0 as u64).encode(s)?;
-Ok(9)
-}
-}
-}
+    #[inline]
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        match self.0 {
+            0..=0xFC => {
+                (self.0 as u8).encode(s)?;
+                Ok(1)
+            }
+            0xFD..=0xFFFF => {
+                s.write_u8(0xFD)?;
+                (self.0 as u16).encode(s)?;
+                Ok(3)
+            }
+            0x10000..=0xFFFFFFFF => {
+                s.write_u8(0xFE)?;
+                (self.0 as u32).encode(s)?;
+                Ok(5)
+            }
+            _ => {
+                s.write_u8(0xFF)?;
+                (self.0 as u64).encode(s)?;
+                Ok(9)
+            }
+        }
+    }
 }
 
 impl Decodable for VarInt {
-#[inline]
-fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-let n = ReadExt::read_u8(&mut d)?;
-match n {
-0xFF => {
-let x = ReadExt::read_u64(&mut d)?;
-if x < 0x100000000 {
-Err(self::Error::NonMinimalVarInt)
-} else {
-Ok(VarInt(x))
-}
-}
-0xFE => {
-let x = ReadExt::read_u32(&mut d)?;
-if x < 0x10000 {
-Err(self::Error::NonMinimalVarInt)
-} else {
-Ok(VarInt(x as u64))
-}
-}
-0xFD => {
-let x = ReadExt::read_u16(&mut d)?;
-if x < 0xFD {
-Err(self::Error::NonMinimalVarInt)
-} else {
-Ok(VarInt(x as u64))
-}
-}
-n => Ok(VarInt(n as u64)),
-}
-}
+    #[inline]
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        let n = ReadExt::read_u8(&mut d)?;
+        match n {
+            0xFF => {
+                let x = ReadExt::read_u64(&mut d)?;
+                if x < 0x100000000 {
+                    Err(self::Error::NonMinimalVarInt)
+                } else {
+                    Ok(VarInt(x))
+                }
+            }
+            0xFE => {
+                let x = ReadExt::read_u32(&mut d)?;
+                if x < 0x10000 {
+                    Err(self::Error::NonMinimalVarInt)
+                } else {
+                    Ok(VarInt(x as u64))
+                }
+            }
+            0xFD => {
+                let x = ReadExt::read_u16(&mut d)?;
+                if x < 0xFD {
+                    Err(self::Error::NonMinimalVarInt)
+                } else {
+                    Ok(VarInt(x as u64))
+                }
+            }
+            n => Ok(VarInt(n as u64)),
+        }
+    }
 }
 
 // Booleans
 impl Encodable for bool {
-#[inline]
-fn encode<S: WriteExt>(&self, mut s: S) -> Result<usize> {
-s.write_bool(*self)?;
-Ok(1)
-}
+    #[inline]
+    fn encode<S: WriteExt>(&self, mut s: S) -> Result<usize> {
+        s.write_bool(*self)?;
+        Ok(1)
+    }
 }
 
 impl Decodable for bool {
-#[inline]
-fn decode<D: io::Read>(mut d: D) -> Result<bool> {
-ReadExt::read_bool(&mut d)
-}
+    #[inline]
+    fn decode<D: io::Read>(mut d: D) -> Result<bool> {
+        ReadExt::read_bool(&mut d)
+    }
 }
 
 // Strings
 impl Encodable for String {
-#[inline]
-fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-let b = self.as_bytes();
-let vi_len = VarInt(b.len() as u64).encode(&mut s)?;
-s.write_slice(&b)?;
-Ok(vi_len + b.len())
-}
+    #[inline]
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let b = self.as_bytes();
+        let vi_len = VarInt(b.len() as u64).encode(&mut s)?;
+        s.write_slice(&b)?;
+        Ok(vi_len + b.len())
+    }
 }
 
 impl Decodable for String {
-#[inline]
-fn decode<D: io::Read>(d: D) -> Result<String> {
-String::from_utf8(Decodable::decode(d)?)
-.map_err(|_| self::Error::ParseFailed("String was not valid UTF8"))
-}
+    #[inline]
+    fn decode<D: io::Read>(d: D) -> Result<String> {
+        String::from_utf8(Decodable::decode(d)?)
+            .map_err(|_| self::Error::ParseFailed("String was not valid UTF8"))
+    }
 }
 
 // Cow<'static, str>
 impl Encodable for Cow<'static, str> {
-#[inline]
-fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-let b = self.as_bytes();
-let vi_len = VarInt(b.len() as u64).encode(&mut s)?;
-s.write_slice(&b)?;
-Ok(vi_len + b.len())
-}
+    #[inline]
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let b = self.as_bytes();
+        let vi_len = VarInt(b.len() as u64).encode(&mut s)?;
+        s.write_slice(&b)?;
+        Ok(vi_len + b.len())
+    }
 }
 
 impl Decodable for Cow<'static, str> {
-#[inline]
-fn decode<D: io::Read>(d: D) -> Result<Cow<'static, str>> {
-String::from_utf8(Decodable::decode(d)?)
-.map_err(|_| self::Error::ParseFailed("String was not valid UTF8"))
-.map(Cow::Owned)
-}
+    #[inline]
+    fn decode<D: io::Read>(d: D) -> Result<Cow<'static, str>> {
+        String::from_utf8(Decodable::decode(d)?)
+            .map_err(|_| self::Error::ParseFailed("String was not valid UTF8"))
+            .map(Cow::Owned)
+    }
 }
 
 // Arrays
 macro_rules! impl_array {
-( $size:expr ) => {
-impl Encodable for [u8; $size] {
-#[inline]
-fn encode<S: WriteExt>(&self, mut s: S) -> Result<usize> {
-s.write_slice(&self[..])?;
-Ok(self.len())
-}
-}
-
-impl Decodable for [u8; $size] {
-#[inline]
-fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-let mut ret = [0; $size];
-d.read_slice(&mut ret)?;
-Ok(ret)
-}
-}
-};
+    ( $size:expr ) => {
+        impl Encodable for [u8; $size] {
+            #[inline]
+            fn encode<S: WriteExt>(&self, mut s: S) -> Result<usize> {
+                s.write_slice(&self[..])?;
+                Ok(self.len())
+            }
+        }
+
+        impl Decodable for [u8; $size] {
+            #[inline]
+            fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+                let mut ret = [0; $size];
+                d.read_slice(&mut ret)?;
+                Ok(ret)
+            }
+        }
+    };
 }
 
 impl_array!(2);
@@ -385,123 +385,123 @@ impl_array!(33);
 // Vectors
 #[macro_export]
 macro_rules! impl_vec {
-($type: ty) => {
-impl Encodable for Vec<$type> {
-#[inline]
-fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-let mut len = 0;
-len += VarInt(self.len() as u64).encode(&mut s)?;
-for c in self.iter() {
-len += c.encode(&mut s)?;
-}
-Ok(len)
-}
-}
-impl Decodable for Vec<$type> {
-#[inline]
-fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-let len = VarInt::decode(&mut d)?.0;
-let mut ret = Vec::with_capacity(len as usize);
-for _ in 0..len {
-ret.push(Decodable::decode(&mut d)?);
-}
-Ok(ret)
-}
-}
-};
+    ($type: ty) => {
+        impl Encodable for Vec<$type> {
+            #[inline]
+            fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+                let mut len = 0;
+                len += VarInt(self.len() as u64).encode(&mut s)?;
+                for c in self.iter() {
+                    len += c.encode(&mut s)?;
+                }
+                Ok(len)
+            }
+        }
+        impl Decodable for Vec<$type> {
+            #[inline]
+            fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+                let len = VarInt::decode(&mut d)?.0;
+                let mut ret = Vec::with_capacity(len as usize);
+                for _ in 0..len {
+                    ret.push(Decodable::decode(&mut d)?);
+                }
+                Ok(ret)
+            }
+        }
+    };
 }
 impl_vec!(bls::Scalar);
 impl_vec!(SocketAddr);
 impl_vec!([u8; 32]);
 
 impl Encodable for IpAddr {
-fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-let mut len = 0;
-match self {
-IpAddr::V4(ip) => {
-let version: u8 = 4;
-len += version.encode(&mut s)?;
-len += ip.octets().encode(s)?;
-}
-IpAddr::V6(ip) => {
-let version: u8 = 6;
-len += version.encode(&mut s)?;
-len += ip.octets().encode(s)?;
-}
-}
-Ok(len)
-}
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        match self {
+            IpAddr::V4(ip) => {
+                let version: u8 = 4;
+                len += version.encode(&mut s)?;
+                len += ip.octets().encode(s)?;
+            }
+            IpAddr::V6(ip) => {
+                let version: u8 = 6;
+                len += version.encode(&mut s)?;
+                len += ip.octets().encode(s)?;
+            }
+        }
+        Ok(len)
+    }
 }
 
 impl Decodable for IpAddr {
-fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-let version: u8 = Decodable::decode(&mut d)?;
-match version {
-4 => {
-let addr: [u8; 4] = Decodable::decode(&mut d)?;
-Ok(IpAddr::from(addr))
-}
-6 => {
-let addr: [u8; 16] = Decodable::decode(&mut d)?;
-Ok(IpAddr::from(addr))
-}
-_ => Err(Error::ParseFailed("couldn't decode IpAddr")),
-}
-}
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        let version: u8 = Decodable::decode(&mut d)?;
+        match version {
+            4 => {
+                let addr: [u8; 4] = Decodable::decode(&mut d)?;
+                Ok(IpAddr::from(addr))
+            }
+            6 => {
+                let addr: [u8; 16] = Decodable::decode(&mut d)?;
+                Ok(IpAddr::from(addr))
+            }
+            _ => Err(Error::ParseFailed("couldn't decode IpAddr")),
+        }
+    }
 }
 
 impl Encodable for SocketAddr {
-fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
-let mut len = 0;
-len += self.ip().encode(&mut s)?;
-len += self.port().encode(s)?;
-Ok(len)
-}
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.ip().encode(&mut s)?;
+        len += self.port().encode(s)?;
+        Ok(len)
+    }
 }
 
 impl Decodable for SocketAddr {
-fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-let ip = Decodable::decode(&mut d)?;
-let port: u16 = Decodable::decode(d)?;
-Ok(SocketAddr::new(ip, port))
-}
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        let ip = Decodable::decode(&mut d)?;
+        let port: u16 = Decodable::decode(d)?;
+        Ok(SocketAddr::new(ip, port))
+    }
 }
 
 pub fn encode_with_size<S: io::Write>(data: &[u8], mut s: S) -> Result<usize> {
-let vi_len = VarInt(data.len() as u64).encode(&mut s)?;
-s.write_slice(&data)?;
-Ok(vi_len + data.len())
+    let vi_len = VarInt(data.len() as u64).encode(&mut s)?;
+    s.write_slice(&data)?;
+    Ok(vi_len + data.len())
 }
 
 impl Encodable for Vec<u8> {
-#[inline]
-fn encode<S: io::Write>(&self, s: S) -> Result<usize> {
-encode_with_size(self, s)
-}
+    #[inline]
+    fn encode<S: io::Write>(&self, s: S) -> Result<usize> {
+        encode_with_size(self, s)
+    }
 }
 
 impl Decodable for Vec<u8> {
-#[inline]
-fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-let len = VarInt::decode(&mut d)?.0 as usize;
-let mut ret = vec![0u8; len];
-d.read_slice(&mut ret)?;
-Ok(ret)
-}
+    #[inline]
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        let len = VarInt::decode(&mut d)?.0 as usize;
+        let mut ret = vec![0u8; len];
+        d.read_slice(&mut ret)?;
+        Ok(ret)
+    }
 }
 
 impl Encodable for Box<[u8]> {
-#[inline]
-fn encode<S: io::Write>(&self, s: S) -> Result<usize> {
-encode_with_size(self, s)
-}
+    #[inline]
+    fn encode<S: io::Write>(&self, s: S) -> Result<usize> {
+        encode_with_size(self, s)
+    }
 }
 
 impl Decodable for Box<[u8]> {
-#[inline]
-fn decode<D: io::Read>(d: D) -> Result<Self> {
-<Vec<u8>>::decode(d).map(From::from)
-}
+    #[inline]
+    fn decode<D: io::Read>(d: D) -> Result<Self> {
+        <Vec<u8>>::decode(d).map(From::from)
+    }
 }
 
 // Tuples
@@ -538,286 +538,286 @@ tuple_encode!(T0, T1, T2, T3, T4, T5, T6, T7);
 
 #[cfg(test)]
 mod tests {
-use super::{deserialize, serialize, Error, Result, VarInt};
-use super::{deserialize_partial, Encodable};
-use crate::endian::{u16_to_array_le, u32_to_array_le, u64_to_array_le};
-use std::io;
-use std::mem::discriminant;
-
-#[test]
-fn serialize_int_test() {
-// bool
-assert_eq!(serialize(&false), vec![0u8]);
-assert_eq!(serialize(&true), vec![1u8]);
-// u8
-assert_eq!(serialize(&1u8), vec![1u8]);
-assert_eq!(serialize(&0u8), vec![0u8]);
-assert_eq!(serialize(&255u8), vec![255u8]);
-// u16
-assert_eq!(serialize(&1u16), vec![1u8, 0]);
-assert_eq!(serialize(&256u16), vec![0u8, 1]);
-assert_eq!(serialize(&5000u16), vec![136u8, 19]);
-// u32
-assert_eq!(serialize(&1u32), vec![1u8, 0, 0, 0]);
-assert_eq!(serialize(&256u32), vec![0u8, 1, 0, 0]);
-assert_eq!(serialize(&5000u32), vec![136u8, 19, 0, 0]);
-assert_eq!(serialize(&500000u32), vec![32u8, 161, 7, 0]);
-assert_eq!(serialize(&168430090u32), vec![10u8, 10, 10, 10]);
-// i32
-assert_eq!(serialize(&-1i32), vec![255u8, 255, 255, 255]);
-assert_eq!(serialize(&-256i32), vec![0u8, 255, 255, 255]);
-assert_eq!(serialize(&-5000i32), vec![120u8, 236, 255, 255]);
-assert_eq!(serialize(&-500000i32), vec![224u8, 94, 248, 255]);
-assert_eq!(serialize(&-168430090i32), vec![246u8, 245, 245, 245]);
-assert_eq!(serialize(&1i32), vec![1u8, 0, 0, 0]);
-assert_eq!(serialize(&256i32), vec![0u8, 1, 0, 0]);
-assert_eq!(serialize(&5000i32), vec![136u8, 19, 0, 0]);
-assert_eq!(serialize(&500000i32), vec![32u8, 161, 7, 0]);
-assert_eq!(serialize(&168430090i32), vec![10u8, 10, 10, 10]);
-// u64
-assert_eq!(serialize(&1u64), vec![1u8, 0, 0, 0, 0, 0, 0, 0]);
-assert_eq!(serialize(&256u64), vec![0u8, 1, 0, 0, 0, 0, 0, 0]);
-assert_eq!(serialize(&5000u64), vec![136u8, 19, 0, 0, 0, 0, 0, 0]);
-assert_eq!(serialize(&500000u64), vec![32u8, 161, 7, 0, 0, 0, 0, 0]);
-assert_eq!(
-serialize(&723401728380766730u64),
-vec![10u8, 10, 10, 10, 10, 10, 10, 10]
-);
-// i64
-assert_eq!(
-serialize(&-1i64),
-vec![255u8, 255, 255, 255, 255, 255, 255, 255]
-);
-assert_eq!(
-serialize(&-256i64),
-vec![0u8, 255, 255, 255, 255, 255, 255, 255]
-);
-assert_eq!(
-serialize(&-5000i64),
-vec![120u8, 236, 255, 255, 255, 255, 255, 255]
-);
-assert_eq!(
-serialize(&-500000i64),
-vec![224u8, 94, 248, 255, 255, 255, 255, 255]
-);
-assert_eq!(
-serialize(&-723401728380766730i64),
-vec![246u8, 245, 245, 245, 245, 245, 245, 245]
-);
-assert_eq!(serialize(&1i64), vec![1u8, 0, 0, 0, 0, 0, 0, 0]);
-assert_eq!(serialize(&256i64), vec![0u8, 1, 0, 0, 0, 0, 0, 0]);
-assert_eq!(serialize(&5000i64), vec![136u8, 19, 0, 0, 0, 0, 0, 0]);
-assert_eq!(serialize(&500000i64), vec![32u8, 161, 7, 0, 0, 0, 0, 0]);
-assert_eq!(
-serialize(&723401728380766730i64),
-vec![10u8, 10, 10, 10, 10, 10, 10, 10]
-);
-}
-
-#[test]
-fn serialize_varint_test() {
-assert_eq!(serialize(&VarInt(10)), vec![10u8]);
-assert_eq!(serialize(&VarInt(0xFC)), vec![0xFCu8]);
-assert_eq!(serialize(&VarInt(0xFD)), vec![0xFDu8, 0xFD, 0]);
-assert_eq!(serialize(&VarInt(0xFFF)), vec![0xFDu8, 0xFF, 0xF]);
-assert_eq!(
-serialize(&VarInt(0xF0F0F0F)),
-vec![0xFEu8, 0xF, 0xF, 0xF, 0xF]
-);
-assert_eq!(
-serialize(&VarInt(0xF0F0F0F0F0E0)),
-vec![0xFFu8, 0xE0, 0xF0, 0xF0, 0xF0, 0xF0, 0xF0, 0, 0]
-);
-assert_eq!(
-test_varint_encode(0xFF, &u64_to_array_le(0x100000000)).unwrap(),
-VarInt(0x100000000)
-);
-assert_eq!(
-test_varint_encode(0xFE, &u64_to_array_le(0x10000)).unwrap(),
-VarInt(0x10000)
-);
-assert_eq!(
-test_varint_encode(0xFD, &u64_to_array_le(0xFD)).unwrap(),
-VarInt(0xFD)
-);
-
-// Test that length calc is working correctly
-test_varint_len(VarInt(0), 1);
-test_varint_len(VarInt(0xFC), 1);
-test_varint_len(VarInt(0xFD), 3);
-test_varint_len(VarInt(0xFFFF), 3);
-test_varint_len(VarInt(0x10000), 5);
-test_varint_len(VarInt(0xFFFFFFFF), 5);
-test_varint_len(VarInt(0xFFFFFFFF + 1), 9);
-test_varint_len(VarInt(u64::max_value()), 9);
-}
-
-fn test_varint_len(varint: VarInt, expected: usize) {
-let mut encoder = io::Cursor::new(vec![]);
-assert_eq!(varint.encode(&mut encoder).unwrap(), expected);
-assert_eq!(varint.len(), expected);
-}
-
-fn test_varint_encode(n: u8, x: &[u8]) -> Result<VarInt> {
-let mut input = [0u8; 9];
-input[0] = n;
-input[1..x.len() + 1].copy_from_slice(x);
-deserialize_partial::<VarInt>(&input).map(|t| t.0)
-}
-
-#[test]
-fn deserialize_nonminimal_vec() {
-// Check the edges for variant int
-assert_eq!(
-discriminant(&test_varint_encode(0xFF, &u64_to_array_le(0x100000000 - 1)).unwrap_err()),
-discriminant(&Error::NonMinimalVarInt)
-);
-assert_eq!(
-discriminant(&test_varint_encode(0xFE, &u32_to_array_le(0x10000 - 1)).unwrap_err()),
-discriminant(&Error::NonMinimalVarInt)
-);
-assert_eq!(
-discriminant(&test_varint_encode(0xFD, &u16_to_array_le(0xFD - 1)).unwrap_err()),
-discriminant(&Error::NonMinimalVarInt)
-);
-
-assert_eq!(
-discriminant(&deserialize::<Vec<u8>>(&[0xfd, 0x00, 0x00]).unwrap_err()),
-discriminant(&Error::NonMinimalVarInt)
-);
-assert_eq!(
-discriminant(&deserialize::<Vec<u8>>(&[0xfd, 0xfc, 0x00]).unwrap_err()),
-discriminant(&Error::NonMinimalVarInt)
-);
-assert_eq!(
-discriminant(&deserialize::<Vec<u8>>(&[0xfd, 0xfc, 0x00]).unwrap_err()),
-discriminant(&Error::NonMinimalVarInt)
-);
-assert_eq!(
-discriminant(&deserialize::<Vec<u8>>(&[0xfe, 0xff, 0x00, 0x00, 0x00]).unwrap_err()),
-discriminant(&Error::NonMinimalVarInt)
-);
-assert_eq!(
-discriminant(&deserialize::<Vec<u8>>(&[0xfe, 0xff, 0xff, 0x00, 0x00]).unwrap_err()),
-discriminant(&Error::NonMinimalVarInt)
-);
-assert_eq!(
-discriminant(
-&deserialize::<Vec<u8>>(&[0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
-.unwrap_err()
-),
-discriminant(&Error::NonMinimalVarInt)
-);
-assert_eq!(
-discriminant(
-&deserialize::<Vec<u8>>(&[0xff, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0x00])
-.unwrap_err()
-),
-discriminant(&Error::NonMinimalVarInt)
-);
-
-let mut vec_256 = vec![0; 259];
-vec_256[0] = 0xfd;
-vec_256[1] = 0x00;
-vec_256[2] = 0x01;
-assert!(deserialize::<Vec<u8>>(&vec_256).is_ok());
-
-let mut vec_253 = vec![0; 256];
-vec_253[0] = 0xfd;
-vec_253[1] = 0xfd;
-vec_253[2] = 0x00;
-assert!(deserialize::<Vec<u8>>(&vec_253).is_ok());
-}
-
-#[test]
-fn serialize_vector_test() {
-assert_eq!(serialize(&vec![1u8, 2, 3]), vec![3u8, 1, 2, 3]);
-// TODO: test vectors of more interesting objects
-}
-
-#[test]
-fn serialize_strbuf_test() {
-assert_eq!(
-serialize(&"Andrew".to_string()),
-vec![6u8, 0x41, 0x6e, 0x64, 0x72, 0x65, 0x77]
-);
-}
-
-#[test]
-fn deserialize_int_test() {
-// bool
-assert!((deserialize(&[58u8, 0]) as Result<bool>).is_err());
-assert_eq!(deserialize(&[58u8]).ok(), Some(true));
-assert_eq!(deserialize(&[1u8]).ok(), Some(true));
-assert_eq!(deserialize(&[0u8]).ok(), Some(false));
-assert!((deserialize(&[0u8, 1]) as Result<bool>).is_err());
-
-// u8
-assert_eq!(deserialize(&[58u8]).ok(), Some(58u8));
-
-// u16
-assert_eq!(deserialize(&[0x01u8, 0x02]).ok(), Some(0x0201u16));
-assert_eq!(deserialize(&[0xABu8, 0xCD]).ok(), Some(0xCDABu16));
-assert_eq!(deserialize(&[0xA0u8, 0x0D]).ok(), Some(0xDA0u16));
-let failure16: Result<u16> = deserialize(&[1u8]);
-assert!(failure16.is_err());
-
-// u32
-assert_eq!(deserialize(&[0xABu8, 0xCD, 0, 0]).ok(), Some(0xCDABu32));
-assert_eq!(
-deserialize(&[0xA0u8, 0x0D, 0xAB, 0xCD]).ok(),
-Some(0xCDAB0DA0u32)
-);
-let failure32: Result<u32> = deserialize(&[1u8, 2, 3]);
-assert!(failure32.is_err());
-// TODO: test negative numbers
-assert_eq!(deserialize(&[0xABu8, 0xCD, 0, 0]).ok(), Some(0xCDABi32));
-assert_eq!(
-deserialize(&[0xA0u8, 0x0D, 0xAB, 0x2D]).ok(),
-Some(0x2DAB0DA0i32)
-);
-let failurei32: Result<i32> = deserialize(&[1u8, 2, 3]);
-assert!(failurei32.is_err());
-
-// u64
-assert_eq!(
-deserialize(&[0xABu8, 0xCD, 0, 0, 0, 0, 0, 0]).ok(),
-Some(0xCDABu64)
-);
-assert_eq!(
-deserialize(&[0xA0u8, 0x0D, 0xAB, 0xCD, 0x99, 0, 0, 0x99]).ok(),
-Some(0x99000099CDAB0DA0u64)
-);
-let failure64: Result<u64> = deserialize(&[1u8, 2, 3, 4, 5, 6, 7]);
-assert!(failure64.is_err());
-// TODO: test negative numbers
-assert_eq!(
-deserialize(&[0xABu8, 0xCD, 0, 0, 0, 0, 0, 0]).ok(),
-Some(0xCDABi64)
-);
-assert_eq!(
-deserialize(&[0xA0u8, 0x0D, 0xAB, 0xCD, 0x99, 0, 0, 0x99]).ok(),
-Some(-0x66ffff663254f260i64)
-);
-let failurei64: Result<i64> = deserialize(&[1u8, 2, 3, 4, 5, 6, 7]);
-assert!(failurei64.is_err());
-}
-
-#[test]
-fn deserialize_vec_test() {
-assert_eq!(deserialize(&[3u8, 2, 3, 4]).ok(), Some(vec![2u8, 3, 4]));
-assert!((deserialize(&[4u8, 2, 3, 4, 5, 6]) as Result<Vec<u8>>).is_err());
-}
-
-#[test]
-fn deserialize_strbuf_test() {
-assert_eq!(
-deserialize(&[6u8, 0x41, 0x6e, 0x64, 0x72, 0x65, 0x77]).ok(),
-Some("Andrew".to_string())
-);
-assert_eq!(
-deserialize(&[6u8, 0x41, 0x6e, 0x64, 0x72, 0x65, 0x77]).ok(),
-Some(::std::borrow::Cow::Borrowed("Andrew"))
-);
-}
+    use super::{deserialize, serialize, Error, Result, VarInt};
+    use super::{deserialize_partial, Encodable};
+    use crate::endian::{u16_to_array_le, u32_to_array_le, u64_to_array_le};
+    use std::io;
+    use std::mem::discriminant;
+
+    #[test]
+    fn serialize_int_test() {
+        // bool
+        assert_eq!(serialize(&false), vec![0u8]);
+        assert_eq!(serialize(&true), vec![1u8]);
+        // u8
+        assert_eq!(serialize(&1u8), vec![1u8]);
+        assert_eq!(serialize(&0u8), vec![0u8]);
+        assert_eq!(serialize(&255u8), vec![255u8]);
+        // u16
+        assert_eq!(serialize(&1u16), vec![1u8, 0]);
+        assert_eq!(serialize(&256u16), vec![0u8, 1]);
+        assert_eq!(serialize(&5000u16), vec![136u8, 19]);
+        // u32
+        assert_eq!(serialize(&1u32), vec![1u8, 0, 0, 0]);
+        assert_eq!(serialize(&256u32), vec![0u8, 1, 0, 0]);
+        assert_eq!(serialize(&5000u32), vec![136u8, 19, 0, 0]);
+        assert_eq!(serialize(&500000u32), vec![32u8, 161, 7, 0]);
+        assert_eq!(serialize(&168430090u32), vec![10u8, 10, 10, 10]);
+        // i32
+        assert_eq!(serialize(&-1i32), vec![255u8, 255, 255, 255]);
+        assert_eq!(serialize(&-256i32), vec![0u8, 255, 255, 255]);
+        assert_eq!(serialize(&-5000i32), vec![120u8, 236, 255, 255]);
+        assert_eq!(serialize(&-500000i32), vec![224u8, 94, 248, 255]);
+        assert_eq!(serialize(&-168430090i32), vec![246u8, 245, 245, 245]);
+        assert_eq!(serialize(&1i32), vec![1u8, 0, 0, 0]);
+        assert_eq!(serialize(&256i32), vec![0u8, 1, 0, 0]);
+        assert_eq!(serialize(&5000i32), vec![136u8, 19, 0, 0]);
+        assert_eq!(serialize(&500000i32), vec![32u8, 161, 7, 0]);
+        assert_eq!(serialize(&168430090i32), vec![10u8, 10, 10, 10]);
+        // u64
+        assert_eq!(serialize(&1u64), vec![1u8, 0, 0, 0, 0, 0, 0, 0]);
+        assert_eq!(serialize(&256u64), vec![0u8, 1, 0, 0, 0, 0, 0, 0]);
+        assert_eq!(serialize(&5000u64), vec![136u8, 19, 0, 0, 0, 0, 0, 0]);
+        assert_eq!(serialize(&500000u64), vec![32u8, 161, 7, 0, 0, 0, 0, 0]);
+        assert_eq!(
+            serialize(&723401728380766730u64),
+            vec![10u8, 10, 10, 10, 10, 10, 10, 10]
+        );
+        // i64
+        assert_eq!(
+            serialize(&-1i64),
+            vec![255u8, 255, 255, 255, 255, 255, 255, 255]
+        );
+        assert_eq!(
+            serialize(&-256i64),
+            vec![0u8, 255, 255, 255, 255, 255, 255, 255]
+        );
+        assert_eq!(
+            serialize(&-5000i64),
+            vec![120u8, 236, 255, 255, 255, 255, 255, 255]
+        );
+        assert_eq!(
+            serialize(&-500000i64),
+            vec![224u8, 94, 248, 255, 255, 255, 255, 255]
+        );
+        assert_eq!(
+            serialize(&-723401728380766730i64),
+            vec![246u8, 245, 245, 245, 245, 245, 245, 245]
+        );
+        assert_eq!(serialize(&1i64), vec![1u8, 0, 0, 0, 0, 0, 0, 0]);
+        assert_eq!(serialize(&256i64), vec![0u8, 1, 0, 0, 0, 0, 0, 0]);
+        assert_eq!(serialize(&5000i64), vec![136u8, 19, 0, 0, 0, 0, 0, 0]);
+        assert_eq!(serialize(&500000i64), vec![32u8, 161, 7, 0, 0, 0, 0, 0]);
+        assert_eq!(
+            serialize(&723401728380766730i64),
+            vec![10u8, 10, 10, 10, 10, 10, 10, 10]
+        );
+    }
+
+    #[test]
+    fn serialize_varint_test() {
+        assert_eq!(serialize(&VarInt(10)), vec![10u8]);
+        assert_eq!(serialize(&VarInt(0xFC)), vec![0xFCu8]);
+        assert_eq!(serialize(&VarInt(0xFD)), vec![0xFDu8, 0xFD, 0]);
+        assert_eq!(serialize(&VarInt(0xFFF)), vec![0xFDu8, 0xFF, 0xF]);
+        assert_eq!(
+            serialize(&VarInt(0xF0F0F0F)),
+            vec![0xFEu8, 0xF, 0xF, 0xF, 0xF]
+        );
+        assert_eq!(
+            serialize(&VarInt(0xF0F0F0F0F0E0)),
+            vec![0xFFu8, 0xE0, 0xF0, 0xF0, 0xF0, 0xF0, 0xF0, 0, 0]
+        );
+        assert_eq!(
+            test_varint_encode(0xFF, &u64_to_array_le(0x100000000)).unwrap(),
+            VarInt(0x100000000)
+        );
+        assert_eq!(
+            test_varint_encode(0xFE, &u64_to_array_le(0x10000)).unwrap(),
+            VarInt(0x10000)
+        );
+        assert_eq!(
+            test_varint_encode(0xFD, &u64_to_array_le(0xFD)).unwrap(),
+            VarInt(0xFD)
+        );
+
+        // Test that length calc is working correctly
+        test_varint_len(VarInt(0), 1);
+        test_varint_len(VarInt(0xFC), 1);
+        test_varint_len(VarInt(0xFD), 3);
+        test_varint_len(VarInt(0xFFFF), 3);
+        test_varint_len(VarInt(0x10000), 5);
+        test_varint_len(VarInt(0xFFFFFFFF), 5);
+        test_varint_len(VarInt(0xFFFFFFFF + 1), 9);
+        test_varint_len(VarInt(u64::max_value()), 9);
+    }
+
+    fn test_varint_len(varint: VarInt, expected: usize) {
+        let mut encoder = io::Cursor::new(vec![]);
+        assert_eq!(varint.encode(&mut encoder).unwrap(), expected);
+        assert_eq!(varint.len(), expected);
+    }
+
+    fn test_varint_encode(n: u8, x: &[u8]) -> Result<VarInt> {
+        let mut input = [0u8; 9];
+        input[0] = n;
+        input[1..x.len() + 1].copy_from_slice(x);
+        deserialize_partial::<VarInt>(&input).map(|t| t.0)
+    }
+
+    #[test]
+    fn deserialize_nonminimal_vec() {
+        // Check the edges for variant int
+        assert_eq!(
+            discriminant(&test_varint_encode(0xFF, &u64_to_array_le(0x100000000 - 1)).unwrap_err()),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+        assert_eq!(
+            discriminant(&test_varint_encode(0xFE, &u32_to_array_le(0x10000 - 1)).unwrap_err()),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+        assert_eq!(
+            discriminant(&test_varint_encode(0xFD, &u16_to_array_le(0xFD - 1)).unwrap_err()),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+
+        assert_eq!(
+            discriminant(&deserialize::<Vec<u8>>(&[0xfd, 0x00, 0x00]).unwrap_err()),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+        assert_eq!(
+            discriminant(&deserialize::<Vec<u8>>(&[0xfd, 0xfc, 0x00]).unwrap_err()),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+        assert_eq!(
+            discriminant(&deserialize::<Vec<u8>>(&[0xfd, 0xfc, 0x00]).unwrap_err()),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+        assert_eq!(
+            discriminant(&deserialize::<Vec<u8>>(&[0xfe, 0xff, 0x00, 0x00, 0x00]).unwrap_err()),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+        assert_eq!(
+            discriminant(&deserialize::<Vec<u8>>(&[0xfe, 0xff, 0xff, 0x00, 0x00]).unwrap_err()),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+        assert_eq!(
+            discriminant(
+                &deserialize::<Vec<u8>>(&[0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])
+                    .unwrap_err()
+            ),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+        assert_eq!(
+            discriminant(
+                &deserialize::<Vec<u8>>(&[0xff, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0x00])
+                    .unwrap_err()
+            ),
+            discriminant(&Error::NonMinimalVarInt)
+        );
+
+        let mut vec_256 = vec![0; 259];
+        vec_256[0] = 0xfd;
+        vec_256[1] = 0x00;
+        vec_256[2] = 0x01;
+        assert!(deserialize::<Vec<u8>>(&vec_256).is_ok());
+
+        let mut vec_253 = vec![0; 256];
+        vec_253[0] = 0xfd;
+        vec_253[1] = 0xfd;
+        vec_253[2] = 0x00;
+        assert!(deserialize::<Vec<u8>>(&vec_253).is_ok());
+    }
+
+    #[test]
+    fn serialize_vector_test() {
+        assert_eq!(serialize(&vec![1u8, 2, 3]), vec![3u8, 1, 2, 3]);
+        // TODO: test vectors of more interesting objects
+    }
+
+    #[test]
+    fn serialize_strbuf_test() {
+        assert_eq!(
+            serialize(&"Andrew".to_string()),
+            vec![6u8, 0x41, 0x6e, 0x64, 0x72, 0x65, 0x77]
+        );
+    }
+
+    #[test]
+    fn deserialize_int_test() {
+        // bool
+        assert!((deserialize(&[58u8, 0]) as Result<bool>).is_err());
+        assert_eq!(deserialize(&[58u8]).ok(), Some(true));
+        assert_eq!(deserialize(&[1u8]).ok(), Some(true));
+        assert_eq!(deserialize(&[0u8]).ok(), Some(false));
+        assert!((deserialize(&[0u8, 1]) as Result<bool>).is_err());
+
+        // u8
+        assert_eq!(deserialize(&[58u8]).ok(), Some(58u8));
+
+        // u16
+        assert_eq!(deserialize(&[0x01u8, 0x02]).ok(), Some(0x0201u16));
+        assert_eq!(deserialize(&[0xABu8, 0xCD]).ok(), Some(0xCDABu16));
+        assert_eq!(deserialize(&[0xA0u8, 0x0D]).ok(), Some(0xDA0u16));
+        let failure16: Result<u16> = deserialize(&[1u8]);
+        assert!(failure16.is_err());
+
+        // u32
+        assert_eq!(deserialize(&[0xABu8, 0xCD, 0, 0]).ok(), Some(0xCDABu32));
+        assert_eq!(
+            deserialize(&[0xA0u8, 0x0D, 0xAB, 0xCD]).ok(),
+            Some(0xCDAB0DA0u32)
+        );
+        let failure32: Result<u32> = deserialize(&[1u8, 2, 3]);
+        assert!(failure32.is_err());
+        // TODO: test negative numbers
+        assert_eq!(deserialize(&[0xABu8, 0xCD, 0, 0]).ok(), Some(0xCDABi32));
+        assert_eq!(
+            deserialize(&[0xA0u8, 0x0D, 0xAB, 0x2D]).ok(),
+            Some(0x2DAB0DA0i32)
+        );
+        let failurei32: Result<i32> = deserialize(&[1u8, 2, 3]);
+        assert!(failurei32.is_err());
+
+        // u64
+        assert_eq!(
+            deserialize(&[0xABu8, 0xCD, 0, 0, 0, 0, 0, 0]).ok(),
+            Some(0xCDABu64)
+        );
+        assert_eq!(
+            deserialize(&[0xA0u8, 0x0D, 0xAB, 0xCD, 0x99, 0, 0, 0x99]).ok(),
+            Some(0x99000099CDAB0DA0u64)
+        );
+        let failure64: Result<u64> = deserialize(&[1u8, 2, 3, 4, 5, 6, 7]);
+        assert!(failure64.is_err());
+        // TODO: test negative numbers
+        assert_eq!(
+            deserialize(&[0xABu8, 0xCD, 0, 0, 0, 0, 0, 0]).ok(),
+            Some(0xCDABi64)
+        );
+        assert_eq!(
+            deserialize(&[0xA0u8, 0x0D, 0xAB, 0xCD, 0x99, 0, 0, 0x99]).ok(),
+            Some(-0x66ffff663254f260i64)
+        );
+        let failurei64: Result<i64> = deserialize(&[1u8, 2, 3, 4, 5, 6, 7]);
+        assert!(failurei64.is_err());
+    }
+
+    #[test]
+    fn deserialize_vec_test() {
+        assert_eq!(deserialize(&[3u8, 2, 3, 4]).ok(), Some(vec![2u8, 3, 4]));
+        assert!((deserialize(&[4u8, 2, 3, 4, 5, 6]) as Result<Vec<u8>>).is_err());
+    }
+
+    #[test]
+    fn deserialize_strbuf_test() {
+        assert_eq!(
+            deserialize(&[6u8, 0x41, 0x6e, 0x64, 0x72, 0x65, 0x77]).ok(),
+            Some("Andrew".to_string())
+        );
+        assert_eq!(
+            deserialize(&[6u8, 0x41, 0x6e, 0x64, 0x72, 0x65, 0x77]).ok(),
+            Some(::std::borrow::Cow::Borrowed("Andrew"))
+        );
+    }
 }

src/system/subscriber.rs

@@ -9,12 +9,12 @@ pub type SubscriptionID = u64;
 
 pub struct Subscription<T> {
     id: SubscriptionID,
-    recv_queue: async_channel::Receiver<Arc<T>>,
+    recv_queue: async_channel::Receiver<T>,
     parent: Arc<Subscriber<T>>,
 }
 
-impl<T> Subscription<T> {
-    pub async fn receive(&self) -> Arc<T> {
+impl<T: Clone> Subscription<T> {
+    pub async fn receive(&self) -> T {
         let message_result = self.recv_queue.recv().await;
 
         match message_result {
@@ -33,10 +33,10 @@ impl<T> Subscription<T> {
 
 // Simple broadcast (publish-subscribe) class
 pub struct Subscriber<T> {
-    subs: Mutex<HashMap<u64, async_channel::Sender<Arc<T>>>>,
+    subs: Mutex<HashMap<u64, async_channel::Sender<T>>>,
 }
 
-impl<T> Subscriber<T> {
+impl<T: Clone> Subscriber<T> {
     pub fn new() -> Arc<Self> {
         Arc::new(Self {
             subs: Mutex::new(HashMap::new()),
@@ -66,7 +66,7 @@ impl<T> Subscriber<T> {
         self.subs.lock().await.remove(&sub_id);
     }
 
-    pub async fn notify(&self, message_result: Arc<T>) {
+    pub async fn notify(&self, message_result: T) {
         for sub in (*self.subs.lock().await).values() {
             match sub.send(message_result.clone()).await {
                 Ok(()) => {}

src/utility.rs

@@ -1,89 +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 ConnectionsMap = std::sync::Arc<
-    async_std::sync::Mutex<HashMap<SocketAddr, async_channel::Sender<net::Message>>>,
->;
-
-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![])
-    }
-}
-
-pub async fn start_connections_process(
-    //slabman: SlabsManagerSafe,
-    stored_addrs: Vec<SocketAddr>,
-    connections: ConnectionsMap,
-    _accept_addr: SocketAddr,
-    _channel_secret: [u8; 32],
-    executor: Arc<Executor<'_>>,
-) -> Vec<Task<()>> {
-    let mut tasks: Vec<Task<()>> = vec![];
-    for _ in 0..10 {
-        let connections_cloned = connections.clone();
-        let stored_addrs_cloned = stored_addrs.clone();
-        //let slabman_cloned = slabman.clone();
-        //let channel_secret = channel_secret.clone();
-        let task = executor.spawn(async move {
-            loop {
-                let addr = stored_addrs_cloned.choose(&mut rand::thread_rng()).unwrap();
-                if !connections_cloned.lock().await.contains_key(addr) {
-                    /*let mut protocol =
-                        ClientProtocol::new(connections_cloned.clone(), slabman_cloned.clone());
-                    protocol
-                        .start(addr.clone(), accept_addr.clone(), &channel_secret)
-                        .await;
-                        */
-                }
-            }
-        });
-        tasks.push(task);
-    }
-    tasks
-}