networking subsystem initial commit (still not working)

Cargo.toml

@@ -27,6 +27,8 @@ rand_xorshift = "0.2"
 blake2s_simd = "0.5"
 bitvec = "0.18"
 bimap = "0.5.2"
+async-trait = "0.1.42"
+multimap = "0.8.2"
 
 hex = "0.4.2"
 num_enum = "0.5.0"
@@ -37,6 +39,8 @@ failure = "0.1.8"
 failure_derive = "0.1.8"
 log = "0.4"
 ctrlc = "3.1.7"
+serde_json = "1.0.61"
+owning_ref = "0.4.1"
 
 smol = "1.2.4"
 futures = "0.3.5"

scripts/jsonrpc_client.py

@@ -7,7 +7,7 @@ def main():
 
     # Example echo method
     payload = {
-        "method": "quit",
+        "method": "get_info",
         "params": [],
         "jsonrpc": "2.0",
         "id": 0,

src/async_serial.rs

@@ -2,11 +2,11 @@ use futures::prelude::*;
 
 use crate::endian;
 use crate::error::{Error, Result};
-use crate::net::net::AsyncTcpStream;
+use crate::net::AsyncTcpStream;
 use crate::serial::VarInt;
 
 impl VarInt {
-    pub async fn encode_async(&self, stream: &mut AsyncTcpStream) -> Result<usize> {
+    pub async fn encode_async<W: AsyncWrite + Unpin>(&self, stream: &mut W) -> Result<usize> {
         match self.0 {
             0..=0xFC => {
                 AsyncWriteExt::write_u8(stream, self.0 as u8).await?;
@@ -30,7 +30,7 @@ impl VarInt {
         }
     }
 
-    pub async fn decode_async(stream: &mut AsyncTcpStream) -> Result<Self> {
+    pub async fn decode_async<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Self> {
         let n = AsyncReadExt::read_u8(stream).await?;
         match n {
             0xFF => {
@@ -65,7 +65,7 @@ impl VarInt {
 macro_rules! async_encoder_fn {
     ($name:ident, $val_type:ty, $writefn:ident) => {
         #[inline]
-        pub async fn $name(stream: &mut AsyncTcpStream, v: $val_type) -> Result<()> {
+        pub async fn $name<W: AsyncWrite + Unpin>(stream: &mut W, v: $val_type) -> Result<()> {
             stream
                 .write_all(&endian::$writefn(v))
                 .await
@@ -76,7 +76,7 @@ macro_rules! async_encoder_fn {
 
 macro_rules! async_decoder_fn {
     ($name:ident, $val_type:ty, $readfn:ident, $byte_len: expr) => {
-        pub async fn $name(stream: &mut AsyncTcpStream) -> Result<$val_type> {
+        pub async fn $name<R: AsyncRead + Unpin>(stream: &mut R) -> 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];
             stream.read_exact(&mut val[..]).await.map_err(Error::Io)?;
@@ -92,7 +92,7 @@ impl AsyncReadExt {
     async_decoder_fn!(read_u32, u32, slice_to_u32_le, 4);
     async_decoder_fn!(read_u16, u16, slice_to_u16_le, 2);
 
-    pub async fn read_u8(stream: &mut AsyncTcpStream) -> Result<u8> {
+    pub async fn read_u8<R: AsyncRead + Unpin>(stream: &mut R) -> Result<u8> {
         let mut slice = [0u8; 1];
         stream.read_exact(&mut slice).await?;
         Ok(slice[0])
@@ -106,7 +106,7 @@ impl AsyncWriteExt {
     async_encoder_fn!(write_u32, u32, u32_to_array_le);
     async_encoder_fn!(write_u16, u16, u16_to_array_le);
 
-    pub async fn write_u8(stream: &mut AsyncTcpStream, v: u8) -> Result<()> {
+    pub async fn write_u8<W: AsyncWrite + Unpin>(stream: &mut W, v: u8) -> Result<()> {
         stream.write_all(&[v]).await.map_err(Error::Io)
     }
 }

src/bin/dfi.rs

@@ -4,11 +4,13 @@ use async_executor::Executor;
 use async_std::sync::Mutex;
 use easy_parallel::Parallel;
 use log::*;
+use serde_json::json;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use sapvi::{ClientProtocol, Result, SeedProtocol, ServerProtocol};
+use sapvi::net;
+use sapvi::{Channel, Result, SeedProtocol, ServerProtocol};
 
 use std::net::TcpListener;
 
@@ -76,15 +78,19 @@ async fn listen(
 }
 
 struct RpcInterface {
+    p2p: Arc<net::P2p>,
+    started: Mutex<bool>,
     quit_send: async_channel::Sender<()>,
     quit_recv: async_channel::Receiver<()>,
 }
 
 impl RpcInterface {
-    fn new() -> Arc<Self> {
+    fn new(p2p: Arc<net::P2p>) -> Arc<Self> {
         let (quit_send, quit_recv) = async_channel::unbounded::<()>();
 
         Arc::new(Self {
+            p2p,
+            started: Mutex::new(false),
             quit_send,
             quit_recv,
         })
@@ -100,6 +106,12 @@ impl RpcInterface {
             Ok(jsonrpc_core::Value::String("Hello World!".into()))
         });
 
+        let self2 = self.clone();
+        io.add_method("get_info", move |_| {
+            let self2 = self2.clone();
+            async move { Ok(json!({"started": *self2.started.lock().await})) }
+        });
+
         let quit_send = self.quit_send.clone();
         io.add_method("quit", move |_| {
             let quit_send = quit_send.clone();
@@ -118,9 +130,37 @@ impl RpcInterface {
         res.set_body(response);
         Ok(res)
     }
+
+    async fn wait_for_quit(self: Arc<Self>) -> Result<()> {
+        Ok(self.quit_recv.recv().await?)
+    }
 }
 
 async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<()> {
+    let p2p = net::P2p::new(options.network_settings);
+
+    let rpc = RpcInterface::new(p2p.clone());
+    let http = listen(
+        executor.clone(),
+        rpc.clone(),
+        Async::<TcpListener>::bind(([127, 0, 0, 1], 8000))?,
+        None,
+    );
+
+    let http_task = executor.spawn(http);
+
+    *rpc.started.lock().await = true;
+
+    p2p.start(executor.clone()).await?;
+
+    rpc.wait_for_quit().await?;
+
+    http_task.cancel().await;
+
+    Ok(())
+}
+
+async fn start2(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<()> {
     let connections = Arc::new(Mutex::new(HashMap::new()));
 
     let stored_addrs = Arc::new(Mutex::new(Vec::new()));
@@ -162,7 +202,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     for i in 0..options.connection_slots {
         debug!("Starting connection slot {}", i);
 
-        let client = ClientProtocol::new(
+        let client = Channel::new(
             connections.clone(),
             accept_addr.clone(),
             stored_addrs.clone(),
@@ -174,7 +214,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     for remote_addr in options.manual_connects {
         debug!("Starting connection (manual) to {}", remote_addr);
 
-        let client = ClientProtocol::new(
+        let client = Channel::new(
             connections.clone(),
             accept_addr.clone(),
             stored_addrs.clone(),
@@ -186,6 +226,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
         client_slots.push(client);
     }
 
+    /*
     let rpc = RpcInterface::new();
     let http = listen(
         executor.clone(),
@@ -199,6 +240,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     rpc.quit_recv.recv().await?;
 
     http_task.cancel().await;
+    */
     match server_task {
         None => {}
         Some(server_task) => {
@@ -209,6 +251,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
 }
 
 struct ProgramOptions {
+    network_settings: net::Settings,
     accept_addr: Option<SocketAddr>,
     seed_addrs: Vec<SocketAddr>,
     manual_connects: Vec<SocketAddr>,
@@ -256,13 +299,26 @@ impl ProgramOptions {
             0
         };
 
-        let log_path = Box::new(if let Some(log_path) = app.value_of("LOG_PATH") {
-            std::path::Path::new(log_path)
-        } else {
-            std::path::Path::new("/tmp/darkfid.log")
-        }.to_path_buf());
+        let log_path = Box::new(
+            if let Some(log_path) = app.value_of("LOG_PATH") {
+                std::path::Path::new(log_path)
+            } else {
+                std::path::Path::new("/tmp/darkfid.log")
+            }
+            .to_path_buf(),
+        );
 
         Ok(ProgramOptions {
+            network_settings: net::Settings {
+                inbound: accept_addr.clone(),
+                outbound_connections: connection_slots,
+                connect_timeout_seconds: 10,
+                channel_handshake_seconds: 2,
+                channel_heartbeat_seconds: 10,
+                external_addr: accept_addr.clone(),
+                peers: manual_connects.clone(),
+                seeds: seed_addrs.clone(),
+            },
             accept_addr,
             seed_addrs,
             manual_connects,

src/error.rs

@@ -32,6 +32,11 @@ pub enum Error {
     VMError(ZKVMError),
     BadContract,
     Groth16Error(bellman::SynthesisError),
+    OperationFailed,
+    ConnectFailed,
+    ConnectTimeout,
+    ChannelStopped,
+    ChannelTimeout,
 }
 
 impl std::error::Error for Error {}
@@ -67,6 +72,11 @@ impl fmt::Display for Error {
             Error::VMError(_) => f.write_str("VM error"),
             Error::BadContract => f.write_str("Contract is poorly defined"),
             Error::Groth16Error(ref err) => write!(f, "groth16 error: {}", err),
+            Error::OperationFailed => f.write_str("Operation failed"),
+            Error::ConnectFailed => f.write_str("Connection failed"),
+            Error::ConnectTimeout => f.write_str("Connection timed out"),
+            Error::ChannelStopped => f.write_str("Channel stopped"),
+            Error::ChannelTimeout => f.write_str("Channel timed out"),
         }
     }
 }

src/lib.rs

@@ -8,14 +8,16 @@ pub mod endian;
 pub mod error;
 pub mod net;
 pub mod serial;
+pub mod system;
 pub mod utility;
 pub mod vm;
 pub mod vm_serial;
 
 pub use crate::bls_extensions::BlsStringConversion;
 pub use crate::error::{Error, Result};
-pub use crate::net::net::{select_event, send_message, sleep};
-pub use crate::net::protocol::client_protocol::ClientProtocol;
+pub use crate::net::messages::{select_event, send_message, sleep};
+pub use crate::net::p2p::P2p;
+pub use crate::net::protocol::client_protocol::Channel;
 pub use crate::net::protocol::seed_protocol::SeedProtocol;
 pub use crate::net::protocol::server_protocol::ServerProtocol;
 pub use crate::serial::{Decodable, Encodable};

src/net/channel.rs

@@ -0,0 +1,120 @@
+use async_std::sync::Mutex;
+use std::sync::atomic::{AtomicBool, Ordering};
+use log::*;
+use futures::FutureExt;
+use futures::io::{ReadHalf, WriteHalf};
+use futures::AsyncReadExt;
+use smol::{Async, Executor};
+use std::future::Future;
+use std::net::{SocketAddr, TcpStream};
+use std::pin::Pin;
+use std::sync::Arc;
+
+use crate::error::{Error, Result};
+use crate::net::messages;
+use crate::net::settings::SettingsPtr;
+use crate::net::message_subscriber::{MessageSubscriberPtr, MessageSubscription, MessageSubscriber};
+use crate::net::utility::clone_net_error;
+use crate::system::{SubscriberPtr, Subscription, Subscriber};
+
+pub type ChannelPtr = Arc<Channel>;
+
+pub struct Channel {
+    reader: Mutex<ReadHalf<Async<TcpStream>>>,
+    writer: Mutex<WriteHalf<Async<TcpStream>>>,
+    address: SocketAddr,
+    message_subscriber: MessageSubscriberPtr,
+    stop_subscriber: SubscriberPtr<Error>,
+    stopped: AtomicBool,
+    settings: SettingsPtr,
+}
+
+impl Channel {
+    pub fn new(stream: Async<TcpStream>, address: SocketAddr, settings: SettingsPtr) -> Arc<Self> {
+        let (reader, writer) = stream.split();
+        let reader = Mutex::new(reader);
+        let writer = Mutex::new(writer);
+        Arc::new(Self {
+            reader,
+            writer,
+            address,
+            message_subscriber: MessageSubscriber::new(),
+            stop_subscriber: Subscriber::new(),
+            stopped: AtomicBool::new(false),
+            settings,
+        })
+    }
+
+    pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
+        executor.spawn(self.receive_loop()).detach();
+    }
+
+    pub async fn send(self: Arc<Self>, message: messages::Message) -> Result<()> {
+        if self.stopped.load(Ordering::Relaxed) {
+            return Err(Error::ChannelStopped);
+        }
+
+        // Catch failure and stop channel, return a net error
+        match messages::send_message(&mut *self.writer.lock().await, message).await {
+            Ok(()) => Ok(()),
+            Err(err) => {
+                error!("Channel error {}, closing {}", err, self.address());
+                self.stop().await;
+                Err(Error::ChannelStopped)
+            }
+        }
+    }
+
+    pub fn address(&self) -> SocketAddr {
+        self.address
+    }
+
+    pub async fn subscribe_msg(self: Arc<Self>, packet_type: messages::PacketType) -> MessageSubscription {
+        self.message_subscriber.clone().subscribe(packet_type).await
+    }
+
+    pub async fn subscribe_stop(self: Arc<Self>) -> Subscription<Error> {
+        self.stop_subscriber.clone().subscribe().await
+    }
+
+    pub async fn stop(&self) {
+        self.stopped.store(false, Ordering::Relaxed);
+        let stop_err = Arc::new(Error::ChannelStopped);
+        self.stop_subscriber.notify(stop_err).await;
+    }
+
+    async fn receive_loop(self: Arc<Self>) -> Result<()> {
+        let stop_sub = self.clone().subscribe_stop().await;
+        let reader = &mut *self.reader.lock().await;
+
+        loop {
+            let message_result = futures::select! {
+                message_result = messages::receive_message(reader).fuse() => {
+                    match message_result {
+                        Ok(message) => Ok(Arc::new(message)),
+                        Err(err) => {
+                            error!("Read error on channel {}", err);
+                            self.stop().await;
+                            Err(Error::ChannelStopped)
+                        }
+                    }
+                }
+                stop_err = stop_sub.receive().fuse() => {
+                    Err(clone_net_error(&*stop_err))
+                }
+            };
+
+            // Save status before using the message
+            let stopped = message_result.is_err();
+
+            // Send result to our subscribers
+            self.message_subscriber.notify(message_result).await;
+
+            // If channel is stopped, timed out or any other error then terminate loop.
+            if stopped {
+                break;
+            }
+        }
+        Ok(())
+    }
+}

src/net/connector.rs

@@ -0,0 +1,31 @@
+use futures::FutureExt;
+use log::*;
+use smol::{Async, Executor};
+use std::net::{SocketAddr, TcpStream};
+use std::sync::Arc;
+
+use crate::error::{Error, Result};
+use crate::net::utility::sleep;
+use crate::net::{Channel, ChannelPtr, SettingsPtr};
+
+pub struct Connector {
+    settings: SettingsPtr,
+}
+
+impl Connector {
+    pub fn new(settings: SettingsPtr) -> Self {
+        Self { settings }
+    }
+
+    pub async fn connect(&self, hostaddr: SocketAddr) -> Result<ChannelPtr> {
+        futures::select! {
+            stream_result = Async::<TcpStream>::connect(hostaddr).fuse() => {
+                match stream_result {
+                    Ok(stream) => Ok(Channel::new(stream, hostaddr, self.settings.clone())),
+                    Err(_) => Err(Error::ConnectFailed)
+                }
+            }
+            _ = sleep(self.settings.connect_timeout_seconds).fuse() => Err(Error::ConnectTimeout)
+        }
+    }
+}

src/net/hosts.rs

@@ -0,0 +1,31 @@
+use std::sync::Arc;
+use rand::seq::SliceRandom;
+use async_std::sync::Mutex;
+use std::net::SocketAddr;
+
+use crate::net::SettingsPtr;
+
+pub type HostsPtr = Arc<Hosts>;
+
+pub struct Hosts {
+    addrs: Mutex<Vec<SocketAddr>>,
+    settings: SettingsPtr
+}
+
+impl Hosts {
+    pub fn new(settings: SettingsPtr) -> Arc<Self> {
+        Arc::new(Self {
+            addrs: Mutex::new(Vec::new()),
+            settings
+        })
+    }
+
+    pub async fn store(&self, addrs: Vec<SocketAddr>) {
+        self.addrs.lock().await.extend(addrs)
+    }
+
+    pub async fn load(&self) -> Option<SocketAddr> {
+        self.addrs.lock().await.choose(&mut rand::thread_rng()).cloned()
+    }
+}
+

src/net/message_subscriber.rs

@@ -0,0 +1,137 @@
+use std::collections::HashMap;
+use rand::Rng;
+use std::sync::Arc;
+use async_std::sync::Mutex;
+
+use crate::error::Result;
+use crate::net::messages::{Message, PacketType};
+use crate::net::utility::clone_net_error;
+
+pub type MessageSubscriberPtr = Arc<MessageSubscriber>;
+
+pub type MessageResult = Result<Arc<Message>>;
+pub type MessageSubscriptionID = u64;
+
+macro_rules! receive_message {
+    ($sub:expr, $message_type:path) => {
+        {
+            let wrapped_message = OwningRef::new($sub.receive().await?);
+
+            wrapped_message.map(|msg|
+                match msg {
+                    $message_type(msg_detail) => {
+                        msg_detail
+                    },
+                    _ => {
+                        panic!("Filter for receive sub invalid!");
+                    }
+            })
+        }
+    };
+}
+
+trait CloneMessageResult {
+    fn clone(&self) -> Self;
+}
+
+impl CloneMessageResult for Result<Arc<Message>> {
+    fn clone(&self) -> Self {
+        match self {
+            Ok(message) => Ok(message.clone()),
+            Err(err) => Err(clone_net_error(err))
+        }
+    }
+}
+
+pub struct MessageSubscription {
+    id: MessageSubscriptionID,
+    filter: PacketType,
+    recv_queue: async_channel::Receiver<MessageResult>,
+    parent: Arc<MessageSubscriber>
+}
+
+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);
+                }
+            }
+        }
+    }
+
+    // Must be called manually since async Drop is not possible in Rust
+    pub async fn unsubscribe(&self) {
+        self.parent.clone().unsubscribe(self.id).await
+    }
+}
+
+pub struct MessageSubscriber {
+    subs: Mutex<HashMap<u64, async_channel::Sender<MessageResult>>>,
+}
+
+impl MessageSubscriber {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self {
+            subs: Mutex::new(HashMap::new()),
+        })
+    }
+
+    pub fn random_id() -> MessageSubscriptionID {
+        let mut rng = rand::thread_rng();
+        rng.gen()
+    }
+
+    pub async fn subscribe(self: Arc<Self>, packet_type: PacketType) -> MessageSubscription {
+        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()
+        }
+    }
+
+    async fn unsubscribe(self: Arc<Self>, sub_id: MessageSubscriptionID) {
+        self.subs.lock().await.remove(&sub_id);
+    }
+
+    pub async fn notify(&self, message_result: Result<Arc<Message>>) {
+        for sub in (*self.subs.lock().await).values() {
+            match sub.send(message_result.clone()).await {
+                Ok(()) => {},
+                Err(err) => {
+                    panic!("Error returned sending message in notify() call! {}", err);
+                }
+            }
+        }
+    }
+}

src/net/net.rs → src/net/messages.rs

@@ -12,17 +12,17 @@ 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};
 
 const MAGIC_BYTES: [u8; 4] = [0xd9, 0xef, 0xb6, 0x7d];
 
-pub type AsyncTcpStream = async_dup::Arc<Async<TcpStream>>;
 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)]
+#[derive(IntoPrimitive, TryFromPrimitive, Copy, Clone, PartialEq, Eq, Hash)]
 #[repr(u8)]
 pub enum PacketType {
     Ping = 0,
@@ -33,6 +33,8 @@ pub enum PacketType {
     Inv = 5,
     GetSlabs = 6,
     Slab = 7,
+    Version = 8,
+    Verack = 9,
 }
 
 pub enum Message {
@@ -44,6 +46,8 @@ pub enum Message {
     Inv(InvMessage),
     GetSlabs(GetSlabsMessage),
     Slab(SlabMessage),
+    Version(VersionMessage),
+    Verack(VerackMessage),
 }
 
 pub struct GetAddrsMessage {}
@@ -66,6 +70,10 @@ pub struct AddrsMessage {
     pub addrs: Vec<SocketAddr>,
 }
 
+pub struct VersionMessage {}
+
+pub struct VerackMessage {}
+
 impl Encodable for GetSlabsMessage {
     fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
         let mut len = 0;
@@ -145,7 +153,56 @@ impl Decodable for AddrsMessage {
     }
 }
 
+impl Encodable for VersionMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        Ok(0)
+    }
+}
+
+impl Decodable for VersionMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {})
+    }
+}
+
+impl Encodable for VerackMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        Ok(0)
+    }
+}
+
+impl Decodable for VerackMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {})
+    }
+}
+
 impl Message {
+    pub fn packet_type(&self) -> PacketType {
+        match self {
+            Message::Ping => 
+                PacketType::Ping,
+            Message::Pong => 
+                PacketType::Pong,
+            Message::GetAddrs(message) => 
+                    PacketType::GetAddrs,
+            Message::Addrs(message) =>
+                    PacketType::Addrs,
+            Message::Sync => 
+                    PacketType::Sync,
+            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 => Ok(Packet {
@@ -200,6 +257,20 @@ impl Message {
                     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,
+                })
+            }
         }
     }
 
@@ -214,6 +285,8 @@ impl Message {
             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)?)),
         }
     }
 
@@ -227,6 +300,8 @@ impl Message {
             Message::Inv(_) => "Inv",
             Message::GetSlabs(_) => "GetSlabs",
             Message::Slab(_) => "Slab",
+            Message::Version(_) => "Version",
+            Message::Verack(_) => "Verack",
         }
     }
 }
@@ -238,7 +313,7 @@ pub struct Packet {
     pub payload: Vec<u8>,
 }
 
-pub async fn read_packet(stream: &mut AsyncTcpStream) -> Result<Packet> {
+pub async fn read_packet<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Packet> {
     // Packets have a 4 byte header of magic digits
     // This is used for network debugging
     let mut magic = [0u8; 4];
@@ -263,7 +338,7 @@ pub async fn read_packet(stream: &mut AsyncTcpStream) -> Result<Packet> {
     Ok(Packet { command, payload })
 }
 
-pub async fn send_packet(stream: &mut AsyncTcpStream, packet: Packet) -> Result<()> {
+pub async fn send_packet<W: AsyncWrite + Unpin>(stream: &mut W, packet: Packet) -> Result<()> {
     stream.write_all(&MAGIC_BYTES).await?;
 
     AsyncWriteExt::write_u8(stream, packet.command as u8).await?;
@@ -278,14 +353,14 @@ pub async fn send_packet(stream: &mut AsyncTcpStream, packet: Packet) -> Result<
     Ok(())
 }
 
-async fn receive_message(stream: &mut AsyncTcpStream) -> Result<Message> {
+pub async fn receive_message<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Message> {
     let packet = read_packet(stream).await?;
     let message = Message::unpack(packet)?;
     debug!("received Message::{}", message.name());
     Ok(message)
 }
 
-pub async fn send_message(stream: &mut AsyncTcpStream, message: Message) -> Result<()> {
+pub async fn send_message<W: AsyncWrite + Unpin>(stream: &mut W, message: Message) -> Result<()> {
     debug!("sending Message::{}", message.name());
     let packet = message.pack()?;
     send_packet(stream, packet).await

src/net/mod.rs

@@ -1,2 +1,26 @@
-pub mod net;
+use smol::Async;
+use std::net::TcpStream;
+
+pub mod channel;
+pub mod connector;
+#[macro_use]
+pub mod message_subscriber;
+pub mod messages;
+pub mod hosts;
+pub mod p2p;
 pub mod protocol;
+pub mod protocols;
+pub mod proxy;
+pub mod sessions;
+pub mod settings;
+pub mod utility;
+
+pub type AsyncTcpStream = async_dup::Arc<Async<TcpStream>>;
+
+pub use channel::{Channel, ChannelPtr};
+pub use connector::Connector;
+pub use message_subscriber::{MessageSubscription, MessageSubscriber};
+pub use hosts::{HostsPtr, Hosts};
+pub use p2p::P2p;
+pub use proxy::Proxy;
+pub use settings::{SettingsPtr, Settings};

src/net/p2p.rs

@@ -0,0 +1,66 @@
+use async_executor::Executor;
+use async_std::sync::Mutex;
+use std::collections::HashMap;
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use crate::error::Result;
+use crate::net::sessions::SeedSession;
+use crate::net::{Channel, ChannelPtr, Hosts, HostsPtr, Connector, Settings, SettingsPtr};
+
+pub type Pending<T> = Mutex<HashMap<SocketAddr, Arc<T>>>;
+
+pub type P2pPtr = Arc<P2p>;
+
+pub struct P2p {
+    pending_connects: Pending<Connector>,
+    pending_channels: Pending<Channel>,
+    hosts: HostsPtr,
+    settings: SettingsPtr,
+}
+
+impl P2p {
+    pub fn new(settings: Settings) -> Arc<Self> {
+        let settings = Arc::new(settings);
+        Arc::new(Self {
+            pending_connects: Mutex::new(HashMap::new()),
+            pending_channels: Mutex::new(HashMap::new()),
+            hosts: Hosts::new(settings.clone()),
+            settings,
+        })
+    }
+
+    /// Invoke startup and seeding sequence. Call from constructing thread.
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        // Start manual connections
+        // Start seed session
+        let seed = SeedSession::new(Arc::downgrade(&self));
+        seed.start(executor.clone()).await?;
+        Ok(())
+    }
+
+    /// Synchronize the blockchain and then begin long running sessions,
+    /// call after start() is invoked.
+    pub fn run() {}
+
+    pub async fn store(self: Arc<Self>, channel: ChannelPtr) {
+        self.pending_channels
+            .lock()
+            .await
+            .insert(channel.address(), channel);
+    }
+    pub async fn remove(self: Arc<Self>, channel: ChannelPtr) {
+        self.pending_channels
+            .lock()
+            .await
+            .remove(&channel.address());
+    }
+
+    pub fn settings(&self) -> SettingsPtr {
+        self.settings.clone()
+    }
+
+    pub fn hosts(&self) -> HostsPtr {
+        self.hosts.clone()
+    }
+}

src/net/protocol/client_protocol.rs

@@ -7,11 +7,11 @@ use std::sync::atomic::AtomicU64;
 use std::sync::Arc;
 
 use crate::error::Result;
-use crate::net::net;
+use crate::net::messages as net;
 use crate::net::protocol::protocol_base;
 use crate::utility::{AddrsStorage, ConnectionsMap};
 
-pub struct ClientProtocol {
+pub struct Channel {
     send_sx: async_channel::Sender<net::Message>,
     send_rx: async_channel::Receiver<net::Message>,
     connections: ConnectionsMap,
@@ -21,7 +21,7 @@ pub struct ClientProtocol {
     stored_addrs: AddrsStorage,
 }
 
-impl ClientProtocol {
+impl Channel {
     pub fn new(
         connections: ConnectionsMap,
         accept_addr: Option<SocketAddr>,

src/net/protocol/protocol_base.rs

@@ -1,7 +1,7 @@
 use log::*;
 use std::sync::atomic::Ordering;
 
-use crate::net::net;
+use crate::net::messages as net;
 use crate::utility::{get_current_time, AddrsStorage, Clock, ConnectionsMap};
 use crate::Result;
 
@@ -104,6 +104,7 @@ pub async fn protocol(
                 .await?;*/
             }
         }
+        _ => {}
     }
     Ok(())
 }

src/net/protocol/seed_protocol.rs

@@ -6,7 +6,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 
 use crate::error::Result;
-use crate::net::net;
+use crate::net::messages as net;
 use crate::net::protocol::protocol_base;
 use crate::utility::{get_current_time, AddrsStorage};
 

src/net/protocol/server_protocol.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 
 //use super::protocol;
 use crate::error::Result;
-use crate::net::net;
+use crate::net::messages as net;
 use crate::net::protocol::protocol_base;
 use crate::utility::{AddrsStorage, ConnectionsMap};
 

src/net/protocols/mod.rs

@@ -0,0 +1,8 @@
+pub mod protocol_version;
+pub mod protocol_ping;
+pub mod protocol_seed;
+
+pub use protocol_version::ProtocolVersion;
+pub use protocol_ping::ProtocolPing;
+pub use protocol_seed::ProtocolSeed;
+

src/net/protocols/protocol_ping.rs

@@ -0,0 +1,51 @@
+use rand::Rng;
+use futures::FutureExt;
+use smol::{Executor, Task};
+use std::sync::Arc;
+
+use crate::error::{Error, Result};
+use crate::net::messages;
+use crate::net::utility::{sleep, clone_net_error};
+use crate::net::{ChannelPtr, SettingsPtr};
+
+pub struct ProtocolPing {
+    channel: ChannelPtr,
+    settings: SettingsPtr,
+}
+
+impl ProtocolPing {
+    pub fn new(channel: ChannelPtr, settings: SettingsPtr) -> Arc<Self> {
+        Arc::new(Self { channel, settings })
+    }
+
+    pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Task<Result<()>> {
+        executor.spawn(self.run_ping_pong())
+    }
+
+    async fn run_ping_pong(self: Arc<Self>) -> Result<()> {
+        let pong_sub = self.channel.clone().subscribe_msg(messages::PacketType::Pong).await;
+
+        loop {
+            // Wait channel_heartbeat amount of time
+            sleep(self.settings.channel_heartbeat_seconds).await;
+
+            // Create a random nonce
+            let _nonce = Self::random_nonce();
+            // TODO: add the nonce after delete other crappy network code
+
+            // Send ping message
+            let ping = messages::Message::Ping;
+            self.channel.clone().send(ping).await?;
+
+            // Wait for pong, check nonce matches
+            let _pong_msg = pong_sub.receive().await?;
+            // TODO: add nonce check here
+        }
+    }
+
+    fn random_nonce() -> u32 {
+        let mut rng = rand::thread_rng();
+        rng.gen()
+    }
+}
+

src/net/protocols/protocol_pong.rs

@@ -0,0 +1,22 @@
+use futures::FutureExt;
+use smol::Executor;
+use std::sync::Arc;
+
+use crate::error::{Error, Result};
+use crate::net::{ChannelPtr, SettingsPtr};
+
+pub struct ProtocolPong {
+    channel: ChannelPtr,
+    settings: SettingsPtr,
+}
+
+impl ProtocolPong {
+    pub fn new(channel: ChannelPtr, settings: SettingsPtr) -> Arc<Self> {
+        Arc::new(Self { channel, settings })
+    }
+
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        Ok(())
+    }
+}
+

src/net/protocols/protocol_seed.rs

@@ -0,0 +1,52 @@
+use futures::FutureExt;
+use smol::Executor;
+use std::sync::Arc;
+use owning_ref::OwningRef;
+
+use crate::error::{Error, Result};
+use crate::net::{ChannelPtr, HostsPtr, SettingsPtr};
+use crate::net::messages;
+
+pub struct ProtocolSeed {
+    channel: ChannelPtr,
+    hosts: HostsPtr,
+    settings: SettingsPtr,
+}
+
+impl ProtocolSeed {
+    pub fn new(channel: ChannelPtr, hosts: HostsPtr, settings: SettingsPtr) -> Arc<Self> {
+        Arc::new(Self { channel, hosts, settings })
+    }
+
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        let addr_sub = self.channel.clone().subscribe_msg(messages::PacketType::Addrs).await;
+
+        // Send own address to the seed server
+        self.send_own_address().await?;
+
+        // Send get address message
+        let get_addr = messages::Message::GetAddrs(messages::GetAddrsMessage {});
+        self.channel.clone().send(get_addr).await?;
+
+        // Receive addresses
+        // TODO: turn this into a macro
+        let addrs_msg = receive_message!(addr_sub, messages::Message::Addrs);
+        self.hosts.store(addrs_msg.addrs.clone()).await;
+
+        Ok(())
+    }
+
+    pub async fn send_own_address(&self) -> Result<()> {
+        match self.settings.external_addr {
+            Some(addr) => {
+                let addr = messages::Message::Addrs(messages::AddrsMessage { addrs: vec![addr] });
+                self.channel.clone().send(addr).await?;
+            },
+            None => {
+                // Do nothing if external address is not configured
+            }
+        }
+        Ok(())
+    }
+}
+

src/net/protocols/protocol_version.rs

@@ -0,0 +1,55 @@
+use futures::FutureExt;
+use smol::Executor;
+use std::sync::Arc;
+
+use crate::error::{Error, Result};
+use crate::net::messages;
+use crate::net::utility::{sleep, clone_net_error};
+use crate::net::{ChannelPtr, SettingsPtr};
+
+pub struct ProtocolVersion {
+    channel: ChannelPtr,
+    settings: SettingsPtr,
+}
+
+impl ProtocolVersion {
+    pub fn new(channel: ChannelPtr, settings: SettingsPtr) -> Arc<Self> {
+        Arc::new(Self { channel, settings })
+    }
+
+    pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        // Start timer
+        // Send version, wait for verack
+        // Wait for version, send verack
+        // Fin.
+        futures::select! {
+            _ = self.clone().exchange_versions(executor).fuse() => Ok(()),
+            _ = sleep(self.settings.channel_handshake_seconds).fuse() => Err(Error::ChannelTimeout)
+        }
+    }
+
+    async fn exchange_versions(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        let send = executor.spawn(self.clone().send_version());
+        let recv = executor.spawn(self.recv_version());
+
+        send.await.and(recv.await)
+    }
+
+    async fn send_version(self: Arc<Self>) -> Result<()> {
+        let version = messages::Message::Version(messages::VersionMessage {});
+
+        self.channel.clone().send(version).await?;
+
+        Ok(())
+    }
+
+    async fn recv_version(self: Arc<Self>) -> Result<()> {
+        let version_sub = self.channel.clone().subscribe_msg(messages::PacketType::Version).await;
+
+        let version_msg = version_sub.receive().await?;
+
+        // Check the message is OK
+
+        Ok(())
+    }
+}

src/net/proxy.rs

@@ -0,0 +1,12 @@
+use smol::{Async, Executor};
+use std::net::{SocketAddr, TcpStream};
+
+pub struct Proxy {
+    stream: Async<TcpStream>,
+}
+
+impl Proxy {
+    pub fn new(stream: Async<TcpStream>) -> Self {
+        Self { stream }
+    }
+}

src/net/sessions/mod.rs

@@ -0,0 +1,5 @@
+pub mod seed_session;
+pub mod session;
+
+pub use seed_session::SeedSession;
+pub use session::Session;

src/net/sessions/seed_session.rs

@@ -0,0 +1,103 @@
+use async_executor::Executor;
+use log::*;
+use std::net::SocketAddr;
+use std::sync::{Arc, Weak};
+
+use crate::error::{Error, Result};
+use crate::net::sessions::Session;
+use crate::net::{ChannelPtr, HostsPtr, Connector, P2p, SettingsPtr};
+use crate::net::protocols::{ProtocolPing, ProtocolSeed};
+
+pub struct SeedSession {
+    p2p: Weak<P2p>
+}
+
+impl SeedSession {
+    pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
+        Arc::new(Self { p2p })
+    }
+
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        let settings = {
+            let p2p = self.p2p.upgrade().unwrap();
+            p2p.settings()
+        };
+
+        // 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(Error::OperationFailed);
+        }
+
+        let mut tasks = Vec::new();
+
+        for seed in settings.seeds.clone() {
+            tasks.push(executor.spawn(self.clone().start_seed(seed, executor.clone())));
+        }
+
+        for task in tasks {
+            // Ignore errors
+            let _ = task.await;
+        }
+
+        // Seed process complete
+        // TODO: check increase count of address
+
+        Ok(())
+    }
+
+    async fn start_seed(self: Arc<Self>, seed: SocketAddr, executor: Arc<Executor<'_>>) -> Result<()> {
+        let (hosts, settings) = {
+            let p2p = self.p2p.upgrade().unwrap();
+            (p2p.hosts(), p2p.settings())
+        };
+
+        let connector = Connector::new(settings.clone());
+        match connector.connect(seed).await {
+            Ok(channel) => {
+                // Blacklist goes here
+
+                info!("Connected seed [{}]", seed);
+
+                self.clone().register_channel(channel.clone(), executor.clone()).await?;
+
+                self.attach_protocols(channel, hosts, settings, executor).await
+            }
+            Err(err) => {
+                info!("Failure contacting seed [{}]: {}", seed, err);
+                Err(err)
+            }
+        }
+    }
+
+    async fn register_channel(self: Arc<Self>, channel: ChannelPtr, executor: Arc<Executor<'_>>) -> Result<()> {
+        let handshake_task = self.perform_handshake_protocols(channel.clone(), executor.clone());
+
+        // start channel
+        channel.start(executor);
+
+        handshake_task.await
+    }
+
+    async fn attach_protocols(self: Arc<Self>, channel: ChannelPtr, hosts: HostsPtr, settings: SettingsPtr, executor: Arc<Executor<'_>>) -> Result<()> {
+        let protocol_ping = ProtocolPing::new(channel.clone(), settings.clone());
+        let ping_task = protocol_ping.start(executor.clone());
+
+        let protocol_seed = ProtocolSeed::new(channel, hosts, settings.clone());
+        protocol_seed.start(executor.clone()).await?;
+
+        // Close the ping task now we finished.
+        // TODO: channel drop should trigger this automatically anyway via the stop signal
+        ping_task.cancel().await;
+
+        Ok(())
+    }
+}
+
+impl Session for SeedSession {
+    fn p2p(&self) -> Arc<P2p> {
+        self.p2p.upgrade().unwrap()
+    }
+}

src/net/sessions/session.rs

@@ -0,0 +1,41 @@
+use async_trait::async_trait;
+use smol::Executor;
+use std::sync::Arc;
+
+use crate::error::Result;
+use crate::net::protocols::ProtocolVersion;
+use crate::net::ChannelPtr;
+use crate::net::p2p::P2pPtr;
+
+async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
+    // Subscribe to stop events
+    let stop_sub = channel.clone().subscribe_stop().await;
+    // Wait for a stop event
+    let _ = stop_sub.receive().await;
+    // Remove channel from p2p
+    p2p.remove(channel).await;
+}
+
+#[async_trait]
+pub trait Session {
+    async fn perform_handshake_protocols(&self, channel: ChannelPtr, executor: Arc<Executor<'_>>) -> Result<()> {
+        let p2p = self.p2p();
+
+        // Perform handshake
+        let protocol_version = ProtocolVersion::new(channel.clone(), p2p.settings());
+        protocol_version.run(executor.clone()).await?;
+
+        // Channel is now initialized
+
+        // Add channel to p2p
+        p2p.clone().store(channel.clone()).await;
+
+        // Subscribe to stop, so can remove from p2p
+        executor.spawn(remove_sub_on_stop(p2p, channel)).detach();
+
+        // Channel is ready for use
+        Ok(())
+    }
+
+    fn p2p(&self) -> P2pPtr;
+}

src/net/settings.rs

@@ -0,0 +1,18 @@
+use std::sync::Arc;
+use std::net::SocketAddr;
+
+pub type SettingsPtr = Arc<Settings>;
+
+#[derive(Clone)]
+pub struct Settings {
+    pub inbound: Option<SocketAddr>,
+    pub outbound_connections: u32,
+
+    pub connect_timeout_seconds: u32,
+    pub channel_handshake_seconds: u32,
+    pub channel_heartbeat_seconds: u32,
+
+    pub external_addr: Option<SocketAddr>,
+    pub peers: Vec<SocketAddr>,
+    pub seeds: Vec<SocketAddr>,
+}

src/net/utility.rs

@@ -0,0 +1,19 @@
+use smol::{Async, Executor, Timer};
+use std::time::Duration;
+
+use crate::error::Error;
+
+pub async fn sleep(seconds: u32) {
+    Timer::after(Duration::from_secs(seconds.into())).await;
+}
+
+pub fn clone_net_error(error: &Error) -> Error {
+    match error {
+        Error::ConnectFailed => Error::ConnectFailed,
+        Error::ConnectTimeout => Error::ConnectTimeout,
+        Error::ChannelStopped => Error::ChannelStopped,
+        Error::ChannelTimeout => Error::ChannelTimeout,
+        _ => Error::OperationFailed
+    }
+}
+

src/utility.rs

@@ -10,7 +10,7 @@ use rand::seq::SliceRandom;
 use smol::{Executor, Task};
 
 //use crate::{net, serial, Channel, ClientProtocol, Result, SlabsManagerSafe};
-use crate::{net::net, serial, Result};
+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>>>,