net: add net3 module

Cargo.toml

@@ -230,6 +230,20 @@ net2 = [
 	"system",
 ]
 
+net3 = [
+	"fxhash",
+	"socket2",
+	"futures-rustls",
+	"fast-socks5",
+	"ed25519-compact",
+	"rcgen",
+	"regex",
+	"rustls-pemfile",
+
+	"util",
+	"system",
+]
+
 crypto = [
 	"bitvec",
 	"blake3",

src/lib.rs

@@ -19,6 +19,9 @@ pub mod net;
 #[cfg(feature = "net2")]
 pub mod net2;
 
+#[cfg(feature = "net3")]
+pub mod net3;
+
 #[cfg(feature = "node")]
 pub mod node;
 

src/net3/acceptor.rs

@@ -0,0 +1,127 @@
+use async_std::stream::StreamExt;
+use std::{net::SocketAddr, sync::Arc};
+
+use futures_rustls::TlsStream;
+use smol::Executor;
+use url::Url;
+
+use crate::{
+    system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
+    Error, Result,
+};
+
+use super::{
+    transport::{TcpTransport, TlsTransport},
+    Channel, ChannelPtr, Transport,
+};
+
+/// Atomic pointer to Acceptor class.
+pub type AcceptorPtr = Arc<Acceptor>;
+
+/// Create inbound socket connections.
+pub struct Acceptor {
+    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
+    task: StoppableTaskPtr,
+}
+
+impl Acceptor {
+    /// Create new Acceptor object.
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self { channel_subscriber: Subscriber::new(), task: StoppableTask::new() })
+    }
+    /// Start accepting inbound socket connections. Creates a listener to start
+    /// listening on a local socket address. Then runs an accept loop in a new
+    /// thread, erroring if a connection problem occurs.
+    pub async fn start(
+        self: Arc<Self>,
+        accept_addr: SocketAddr,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        self.accept(accept_addr, executor);
+        Ok(())
+    }
+
+    /// Stop accepting inbound socket connections.
+    pub async fn stop(&self) {
+        // Send stop signal
+        self.task.stop().await;
+    }
+
+    /// Start receiving network messages.
+    pub async fn subscribe(self: Arc<Self>) -> Subscription<Result<ChannelPtr>> {
+        self.channel_subscriber.clone().subscribe().await
+    }
+
+    /// Run the accept loop in a new thread and error if a connection problem
+    /// occurs.
+    fn accept(self: Arc<Self>, accept_addr: SocketAddr, executor: Arc<Executor<'_>>) {
+        self.task.clone().start(
+            self.clone().run_accept_loop(accept_addr),
+            |result| self.handle_stop(result),
+            Error::ServiceStopped,
+            executor,
+        );
+    }
+
+    /// Run the accept loop.
+    async fn run_accept_loop(self: Arc<Self>, accept_addr: SocketAddr) -> Result<()> {
+        let mut url = Url::parse(&accept_addr.to_string())?;
+        url.set_host(Some("tcp"))?;
+
+        match url.scheme() {
+            "tcp" => {
+                let transport = TcpTransport::new(None, 1024);
+                let listener = transport.listen_on(url).unwrap().await.unwrap();
+                let mut incoming = listener.incoming();
+                while let Some(stream) = incoming.next().await {
+                    let stream = stream.unwrap();
+                    let peer_addr = stream.peer_addr().unwrap();
+                    let channel = Channel::new(Box::new(stream), peer_addr).await;
+                }
+            }
+            "tls" => {
+                let transport = TlsTransport::new(None, 1024);
+                let (acceptor, listener) = transport.listen_on(url).unwrap().await.unwrap();
+                let mut incoming = listener.incoming();
+                while let Some(stream) = incoming.next().await {
+                    let stream = stream.unwrap();
+                    let peer_addr = stream.peer_addr().unwrap();
+                    let mut stream = acceptor.accept(stream).await.unwrap();
+                    let channel =
+                        Channel::new(Box::new(TlsStream::Server(stream)), peer_addr).await;
+                }
+            }
+            _ => unimplemented!(),
+        }
+        Ok(())
+    }
+
+    /// Handles network errors. Panics if error passes silently, otherwise
+    /// broadcasts the error.
+    async fn handle_stop(self: Arc<Self>, result: Result<()>) {
+        match result {
+            Ok(()) => panic!("Acceptor task should never complete without error status"),
+            Err(err) => {
+                // Send this error to all channel subscribers
+                let result = Err(err);
+                self.channel_subscriber.notify(result).await;
+            }
+        }
+    }
+
+    ///// Single attempt to accept an incoming connection. Stops after one
+    ///// attempt.
+    //async fn tick_accept(&self, listener: &TcpListener) -> Result<ChannelPtr> {
+    //    let (stream, peer_addr) = match listener.accept().await {
+    //        Ok((s, a)) => (s, a),
+    //        Err(err) => {
+    //            error!("Error listening for connections: {}", err);
+    //            return Err(Error::ServiceStopped)
+    //        }
+    //    };
+    //    info!("Accepted client: {}", peer_addr);
+
+    //    let channel = Channel::new(Box::new(stream), peer_addr).await;
+    //    Ok(channel)
+    //}
+}

src/net3/channel.rs

@@ -0,0 +1,304 @@
+use async_std::{net::TcpStream, sync::Mutex};
+use std::{
+    net::SocketAddr,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+};
+
+use futures::{
+    io::{ReadHalf, WriteHalf},
+    AsyncRead, AsyncReadExt, AsyncWrite,
+};
+use futures_rustls::TlsStream;
+use log::{debug, error, info};
+use rand::Rng;
+use serde_json::json;
+use smol::Executor;
+
+use crate::{
+    system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
+    Error, Result,
+};
+
+use super::{
+    message,
+    message_subscriber::{MessageSubscription, MessageSubsystem},
+};
+
+/// Atomic pointer to async channel.
+pub type ChannelPtr = Arc<Channel>;
+
+struct ChannelInfo {
+    random_id: u32,
+    last_msg: String,
+    last_status: String,
+    // Message log which is cleared on querying get_info
+    log: Mutex<Vec<(String, String)>>,
+}
+
+impl ChannelInfo {
+    fn new() -> Self {
+        Self {
+            random_id: rand::thread_rng().gen(),
+            last_msg: String::new(),
+            last_status: String::new(),
+            log: Mutex::new(Vec::new()),
+        }
+    }
+
+    async fn get_info(&self) -> serde_json::Value {
+        let result = json!({
+            "random_id": self.random_id,
+            "last_msg": self.last_msg,
+            "last_status": self.last_status,
+            "log": self.log.lock().await.clone(),
+        });
+        self.log.lock().await.clear();
+        result
+    }
+}
+
+pub trait Stream: AsyncWrite + AsyncRead + Unpin + Send + Sync {}
+
+impl Stream for TcpStream {}
+impl<T: Stream> Stream for TlsStream<T> {}
+
+/// Async channel for communication between nodes.
+pub struct Channel {
+    reader: Mutex<ReadHalf<Box<dyn Stream>>>,
+    writer: Mutex<WriteHalf<Box<dyn Stream>>>,
+    address: SocketAddr,
+    message_subsystem: MessageSubsystem,
+    stop_subscriber: SubscriberPtr<Error>,
+    receive_task: StoppableTaskPtr,
+    stopped: AtomicBool,
+    info: Mutex<ChannelInfo>,
+}
+
+impl Channel {
+    /// Sets up a new channel. Creates a reader and writer TCP stream and
+    /// summons the message subscriber subsystem. Performs a network
+    /// handshake on the subsystem dispatchers.
+    pub async fn new(stream: Box<dyn Stream>, 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_subsystem,
+            stop_subscriber: Subscriber::new(),
+            receive_task: StoppableTask::new(),
+            stopped: AtomicBool::new(false),
+            info: Mutex::new(ChannelInfo::new()),
+        })
+    }
+
+    pub async fn get_info(&self) -> serde_json::Value {
+        self.info.lock().await.get_info().await
+    }
+
+    /// Starts the channel. Runs a receive loop to start receiving messages or
+    /// handles a network failure.
+    pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
+        debug!(target: "net", "Channel::start() [START, address={}]", self.address());
+        let self2 = self.clone();
+        self.receive_task.clone().start(
+            self.clone().main_receive_loop(),
+            // Ignore stop handler
+            |result| self2.handle_stop(result),
+            Error::ServiceStopped,
+            executor,
+        );
+        debug!(target: "net", "Channel::start() [END, address={}]", self.address());
+    }
+
+    /// Stops the channel. Steps through each component of the channel
+    /// connection and sends a stop signal. Notifies all subscribers that
+    /// the channel has been closed.
+    pub async fn stop(&self) {
+        debug!(target: "net", "Channel::stop() [START, address={}]", self.address());
+        assert!(!self.stopped.load(Ordering::Relaxed));
+        // Changes memory ordering to relaxed. We don't need strict thread locking here.
+        self.stopped.store(false, Ordering::Relaxed);
+        self.stop_subscriber.notify(Error::ChannelStopped).await;
+        self.receive_task.stop().await;
+        self.message_subsystem.trigger_error(Error::ChannelStopped).await;
+        debug!(target: "net", "Channel::stop() [END, address={}]", self.address());
+    }
+
+    /// Creates a subscription to a stopped signal.
+    pub async fn subscribe_stop(&self) -> Subscription<Error> {
+        debug!(target: "net",
+         "Channel::subscribe_stop() [START, address={}]",
+         self.address()
+        );
+        // TODO: this should check the stopped status
+        // Call to receive should return ChannelStopped on newly created sub
+        let sub = self.stop_subscriber.clone().subscribe().await;
+        debug!(target: "net",
+         "Channel::subscribe_stop() [END, address={}]",
+         self.address()
+        );
+        sub
+    }
+
+    /// Sends a message across a channel. Calls function 'send_message' that
+    /// creates a new payload and sends it over the TCP connection as a
+    /// packet. Returns an error if something goes wrong.
+    pub async fn send<M: message::Message>(&self, message: M) -> Result<()> {
+        debug!(target: "net",
+         "Channel::send() [START, command={:?}, address={}]",
+         M::name(),
+         self.address()
+        );
+        if self.stopped.load(Ordering::Relaxed) {
+            return Err(Error::ChannelStopped)
+        }
+
+        // Catch failure and stop channel, return a net error
+        let result = match self.send_message(message).await {
+            Ok(()) => Ok(()),
+            Err(err) => {
+                error!("Channel send error for [{}]: {}", self.address(), err);
+                self.stop().await;
+                Err(Error::ChannelStopped)
+            }
+        };
+
+        debug!(target: "net",
+         "Channel::send() [END, command={:?}, address={}]",
+         M::name(),
+         self.address()
+        );
+        {
+            let info = &mut *self.info.lock().await;
+            info.last_msg = M::name().to_string();
+            info.last_status = "sent".to_string();
+        }
+
+        result
+    }
+
+    /// Implements send message functionality. Creates a new payload and encodes
+    /// it. Then creates a message packet- the base type of the network- and
+    /// copies the payload into it. Then we send the packet over the TCP
+    /// stream.
+    async fn send_message<M: message::Message>(&self, message: M) -> Result<()> {
+        let mut payload = Vec::new();
+        message.encode(&mut payload)?;
+        let packet = message::Packet { command: String::from(M::name()), payload };
+
+        {
+            let info = &mut *self.info.lock().await;
+            info.log.lock().await.push(("send".to_string(), packet.command.clone()));
+        }
+
+        let stream = &mut *self.writer.lock().await;
+        message::send_packet(stream, packet).await
+    }
+
+    /// Subscribe to a messages on the message subsystem.
+    pub async fn subscribe_msg<M: message::Message>(&self) -> Result<MessageSubscription<M>> {
+        debug!(target: "net",
+         "Channel::subscribe_msg() [START, command={:?}, address={}]",
+         M::name(),
+         self.address()
+        );
+        let sub = self.message_subsystem.subscribe::<M>().await;
+        debug!(target: "net",
+         "Channel::subscribe_msg() [END, command={:?}, address={}]",
+         M::name(),
+         self.address()
+        );
+        sub
+    }
+
+    /// Return the local socket address.
+    pub fn address(&self) -> SocketAddr {
+        self.address
+    }
+
+    /// End of file error. Triggered when unexpected end of file occurs.
+    fn is_eof_error(err: Error) -> bool {
+        match err {
+            Error::Io(io_err) => io_err == std::io::ErrorKind::UnexpectedEof,
+            _ => false,
+        }
+    }
+
+    /// Perform network handshake for message subsystem dispatchers.
+    async fn setup_dispatchers(message_subsystem: &MessageSubsystem) {
+        message_subsystem.add_dispatch::<message::VersionMessage>().await;
+        message_subsystem.add_dispatch::<message::VerackMessage>().await;
+        message_subsystem.add_dispatch::<message::PingMessage>().await;
+        message_subsystem.add_dispatch::<message::PongMessage>().await;
+        message_subsystem.add_dispatch::<message::GetAddrsMessage>().await;
+        message_subsystem.add_dispatch::<message::AddrsMessage>().await;
+    }
+
+    /// Convenience function that returns the Message Subsystem.
+    pub fn get_message_subsystem(&self) -> &MessageSubsystem {
+        &self.message_subsystem
+    }
+
+    /// Run the receive loop. Start receiving messages or handle network
+    /// failure.
+    async fn main_receive_loop(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net",
+         "Channel::receive_loop() [START, address={}]",
+         self.address()
+        );
+
+        let reader = &mut *self.reader.lock().await;
+
+        loop {
+            let packet = match message::read_packet(reader).await {
+                Ok(packet) => packet,
+                Err(err) => {
+                    if Self::is_eof_error(err.clone()) {
+                        info!("Channel {:?} disconnected", self.address());
+                    } else {
+                        error!("Read error on channel: {}", err);
+                    }
+                    debug!(target: "net",
+                     "Channel::receive_loop() stopping channel {:?}",
+                     self.address()
+                    );
+                    self.stop().await;
+                    return Err(Error::ChannelStopped)
+                }
+            };
+            {
+                let info = &mut *self.info.lock().await;
+                info.last_msg = packet.command.clone();
+                info.last_status = "recv".to_string();
+                info.log.lock().await.push(("recv".to_string(), packet.command.clone()));
+            }
+
+            // Send result to our subscribers
+            self.message_subsystem.notify(&packet.command, packet.payload).await;
+        }
+    }
+
+    /// Handle network errors. Panic if error passes silently, otherwise
+    /// broadcast the error.
+    async fn handle_stop(self: Arc<Self>, result: Result<()>) {
+        debug!(target: "net", "Channel::handle_stop() [START, address={}]", self.address());
+        match result {
+            Ok(()) => panic!("Channel task should never complete without error status"),
+            Err(err) => {
+                // Send this error to all channel subscribers
+                self.message_subsystem.trigger_error(err).await;
+            }
+        }
+        debug!(target: "net", "Channel::handle_stop() [END, address={}]", self.address());
+    }
+}

src/net3/connector.rs

@@ -0,0 +1,47 @@
+use async_std::{future::timeout, net::TcpStream};
+use std::{net::SocketAddr, time::Duration};
+use url::Url;
+
+use crate::{Error, Result};
+
+use super::{
+    transport::{TcpTransport, TlsTransport},
+    Channel, ChannelPtr, SettingsPtr, Transport,
+};
+
+/// Create outbound socket connections.
+pub struct Connector {
+    settings: SettingsPtr,
+}
+
+impl Connector {
+    /// Create a new connector with default network settings.
+    pub fn new(settings: SettingsPtr) -> Self {
+        Self { settings }
+    }
+
+    /// Establish an outbound connection.
+    pub async fn connect(&self, hosturl: SocketAddr) -> Result<ChannelPtr> {
+        let mut url = Url::parse(&hosturl.to_string())?;
+        url.set_host(Some("tcp"))?;
+        let result =
+            timeout(Duration::from_secs(self.settings.connect_timeout_seconds.into()), async {
+                match url.scheme() {
+                    "tcp" => {
+                        let transport = TcpTransport::new(None, 1024);
+                        let stream = transport.dial(url).unwrap().await.unwrap();
+                        Channel::new(Box::new(stream), hosturl).await
+                    }
+                    "tls" => {
+                        let transport = TlsTransport::new(None, 1024);
+                        let stream = transport.dial(url).unwrap().await.unwrap();
+                        Channel::new(Box::new(stream), hosturl).await
+                    }
+                    _ => unimplemented!(),
+                }
+            })
+            .await
+            .unwrap();
+        Ok(result)
+    }
+}

src/net3/hosts.rs

@@ -0,0 +1,42 @@
+use async_std::sync::Mutex;
+use std::{net::SocketAddr, sync::Arc};
+
+use fxhash::FxHashSet;
+
+/// Pointer to hosts class.
+pub type HostsPtr = Arc<Hosts>;
+
+/// Manages a store of network addresses.
+pub struct Hosts {
+    addrs: Mutex<Vec<SocketAddr>>,
+}
+
+impl Hosts {
+    /// Create a new host list.
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self { addrs: Mutex::new(Vec::new()) })
+    }
+
+    /// Checks if a host address is in the host list.
+    async fn contains(&self, addrs: &[SocketAddr]) -> bool {
+        let a_set: FxHashSet<_> = addrs.iter().copied().collect();
+        self.addrs.lock().await.iter().any(|item| a_set.contains(item))
+    }
+
+    /// Add a new host to the host list.
+    pub async fn store(&self, addrs: Vec<SocketAddr>) {
+        if !self.contains(&addrs).await {
+            self.addrs.lock().await.extend(addrs)
+        }
+    }
+
+    /// Return the list of hosts.
+    pub async fn load_all(&self) -> Vec<SocketAddr> {
+        self.addrs.lock().await.clone()
+    }
+
+    /// Check if the host list is empty.
+    pub async fn is_empty(&self) -> bool {
+        self.addrs.lock().await.is_empty()
+    }
+}

src/net3/message.rs

@@ -0,0 +1,221 @@
+use std::{io, net::SocketAddr};
+
+use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use log::debug;
+
+use crate::{
+    util::serial::{Decodable, Encodable, VarInt},
+    Error, Result,
+};
+
+const MAGIC_BYTES: [u8; 4] = [0xd9, 0xef, 0xb6, 0x7d];
+
+/// Generic message template.
+pub trait Message: 'static + Encodable + Decodable + Send + Sync {
+    fn name() -> &'static str;
+}
+
+/// Outbound keep-alive message.
+pub struct PingMessage {
+    pub nonce: u32,
+}
+
+/// Inbound keep-alive message.
+pub struct PongMessage {
+    pub nonce: u32,
+}
+
+/// Requests address of outbound connection.
+pub struct GetAddrsMessage {}
+
+/// Sends address information to inbound connection. Response to GetAddrs
+/// message.
+pub struct AddrsMessage {
+    pub addrs: Vec<SocketAddr>,
+}
+
+/// Requests version information of outbound connection.
+pub struct VersionMessage {}
+
+/// Sends version information to inbound connection. Response to VersionMessage.
+pub struct VerackMessage {}
+
+impl Message for PingMessage {
+    fn name() -> &'static str {
+        "ping"
+    }
+}
+
+impl Message for PongMessage {
+    fn name() -> &'static str {
+        "pong"
+    }
+}
+
+impl Message for GetAddrsMessage {
+    fn name() -> &'static str {
+        "getaddr"
+    }
+}
+
+impl Message for AddrsMessage {
+    fn name() -> &'static str {
+        "addr"
+    }
+}
+
+impl Message for VersionMessage {
+    fn name() -> &'static str {
+        "version"
+    }
+}
+
+impl Message for VerackMessage {
+    fn name() -> &'static str {
+        "verack"
+    }
+}
+
+impl Encodable for PingMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.nonce.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for PingMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self { nonce: Decodable::decode(&mut d)? })
+    }
+}
+
+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 Decodable for PongMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self { nonce: Decodable::decode(&mut d)? })
+    }
+}
+
+impl Encodable for GetAddrsMessage {
+    fn encode<S: io::Write>(&self, mut _s: S) -> Result<usize> {
+        let len = 0;
+        Ok(len)
+    }
+}
+
+impl Decodable for GetAddrsMessage {
+    fn decode<D: io::Read>(mut _d: D) -> Result<Self> {
+        Ok(Self {})
+    }
+}
+
+impl Encodable for AddrsMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.addrs.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for AddrsMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self { addrs: Decodable::decode(&mut d)? })
+    }
+}
+
+impl Encodable for VersionMessage {
+    fn encode<S: io::Write>(&self, _s: S) -> Result<usize> {
+        Ok(0)
+    }
+}
+
+impl Decodable for VersionMessage {
+    fn decode<D: io::Read>(_d: D) -> Result<Self> {
+        Ok(Self {})
+    }
+}
+
+impl Encodable for VerackMessage {
+    fn encode<S: io::Write>(&self, _s: S) -> Result<usize> {
+        Ok(0)
+    }
+}
+
+impl Decodable for VerackMessage {
+    fn decode<D: io::Read>(_d: D) -> Result<Self> {
+        Ok(Self {})
+    }
+}
+
+/// Packets are the base type read from the network. Converted to messages and
+/// passed to event loop.
+pub struct Packet {
+    pub command: String,
+    pub payload: Vec<u8>,
+}
+
+/// Reads and decodes an inbound payload.
+pub async fn read_packet<R: AsyncRead + Unpin + Sized>(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];
+    debug!(target: "net", "reading magic...");
+    stream.read_exact(&mut magic).await?;
+    debug!(target: "net", "read magic {:?}", magic);
+    if magic != MAGIC_BYTES {
+        return Err(Error::MalformedPacket)
+    }
+
+    // The type of the message
+    let command_len = VarInt::decode_async(stream).await?.0 as usize;
+    let mut cmd = vec![0u8; command_len];
+    if command_len > 0 {
+        stream.read_exact(&mut cmd).await?;
+    }
+    let cmd = String::from_utf8(cmd)?;
+    debug!(target: "net", "read command: {}", cmd);
+
+    let payload_len = VarInt::decode_async(stream).await?.0 as usize;
+
+    // The message-dependent data (see message types)
+    let mut payload = vec![0u8; payload_len];
+    if payload_len > 0 {
+        stream.read_exact(&mut payload).await?;
+    }
+    debug!(target: "net", "read payload {} bytes", payload_len);
+
+    Ok(Packet { command: cmd, payload })
+}
+
+/// Sends an outbound packet by writing data to TCP stream.
+pub async fn send_packet<W: AsyncWrite + Unpin + Sized>(
+    stream: &mut W,
+    packet: Packet,
+) -> Result<()> {
+    debug!(target: "net", "sending magic...");
+    stream.write_all(&MAGIC_BYTES).await?;
+    debug!(target: "net", "sent magic...");
+
+    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).encode_async(stream).await?;
+
+    if !packet.payload.is_empty() {
+        stream.write_all(&packet.payload).await?;
+    }
+    debug!(target: "net", "sent payload {} bytes", packet.payload.len() as u64);
+
+    Ok(())
+}

src/net3/message_subscriber.rs

@@ -0,0 +1,308 @@
+use async_std::sync::Mutex;
+use std::{any::Any, io, io::Cursor, sync::Arc};
+
+use async_trait::async_trait;
+use fxhash::FxHashMap;
+use log::{debug, error, warn};
+use rand::Rng;
+
+use crate::{
+    util::serial::{Decodable, Encodable},
+    Error, Result,
+};
+
+use super::message::Message;
+
+/// 64bit identifier for message subscription.
+pub type MessageSubscriptionId = u64;
+type MessageResult<M> = Result<Arc<M>>;
+
+/// Handles message subscriptions through a subscription ID and a receiver
+/// channel.
+pub struct MessageSubscription<M: Message> {
+    id: MessageSubscriptionId,
+    recv_queue: async_channel::Receiver<MessageResult<M>>,
+    parent: Arc<MessageDispatcher<M>>,
+}
+
+impl<M: Message> MessageSubscription<M> {
+    /// Start receiving messages.
+    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);
+            }
+        }
+    }
+
+    /// Unsubscribe from a message subscription. Must be called manually.
+    pub async fn unsubscribe(&self) {
+        self.parent.clone().unsubscribe(self.id).await
+    }
+}
+
+#[async_trait]
+/// Generic interface for message dispatcher.
+trait MessageDispatcherInterface: Send + Sync {
+    async fn trigger(&self, payload: Vec<u8>);
+
+    async fn trigger_error(&self, err: Error);
+
+    fn as_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync>;
+}
+
+/// Maintains a list of active subscribers and handles sending messages across
+/// subscriptions.
+struct MessageDispatcher<M: Message> {
+    subs: Mutex<FxHashMap<MessageSubscriptionId, async_channel::Sender<MessageResult<M>>>>,
+}
+
+impl<M: Message> MessageDispatcher<M> {
+    /// Create a new message dispatcher.
+    fn new() -> Self {
+        MessageDispatcher { subs: Mutex::new(FxHashMap::default()) }
+    }
+
+    /// Create a random ID.
+    fn random_id() -> MessageSubscriptionId {
+        let mut rng = rand::thread_rng();
+        rng.gen()
+    }
+
+    /// Subscribe to a channel. Assigns a new ID and adds it to the list of
+    /// subscribers.
+    pub async fn subscribe(self: Arc<Self>) -> MessageSubscription<M> {
+        let (sender, recvr) = async_channel::unbounded();
+        let sub_id = Self::random_id();
+        self.subs.lock().await.insert(sub_id, sender);
+
+        MessageSubscription { id: sub_id, recv_queue: recvr, parent: self }
+    }
+
+    /// Unsubcribe from a channel. Removes the associated ID from the subscriber
+    /// list.
+    async fn unsubscribe(&self, sub_id: MessageSubscriptionId) {
+        self.subs.lock().await.remove(&sub_id);
+    }
+
+    /// Send a message to all subscriber channels. Automatically clear inactive
+    /// channels.
+    async fn trigger_all(&self, message: MessageResult<M>) {
+        debug!(
+            target: "net",
+            "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.clone()).await {
+                Ok(()) => {}
+                Err(_err) => {
+                    // Automatically clean out closed channels
+                    garbage_ids.push(*sub_id);
+                    // panic!("Error returned sending message in notify() call!
+                    // {}", err);
+                }
+            }
+        }
+
+        self.collect_garbage(garbage_ids).await;
+
+        debug!(
+            target: "net",
+            "MessageDispatcher<M={}>::trigger_all({}) [END, subs={}]",
+            M::name(),
+            if message.is_ok() { "msg" } else { "err" },
+            self.subs.lock().await.len()
+        );
+    }
+
+    /// Remove inactive channels.
+    async fn collect_garbage(&self, ids: Vec<MessageSubscriptionId>) {
+        let mut subs = self.subs.lock().await;
+        for id in &ids {
+            subs.remove(id);
+        }
+    }
+}
+
+#[async_trait]
+// Local implementation of the Message Dispatcher Interface.
+impl<M: Message> MessageDispatcherInterface for MessageDispatcher<M> {
+    /// Deserialize data into a message type.
+    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);
+            }
+        }
+    }
+
+    /// Sends a message to all subscriber channels. Clears any inactive
+    /// channels.
+    async fn trigger_error(&self, err: Error) {
+        self.trigger_all(Err(err)).await;
+    }
+
+    /// Converts to Any trait. Enables the dynamic modification of static types.
+    fn as_any(self: Arc<Self>) -> Arc<dyn Any + Send + Sync> {
+        self
+    }
+}
+
+/// Publish/subscribe class that can dispatch any kind of message to a
+/// list of dispatchers.
+pub struct MessageSubsystem {
+    dispatchers: Mutex<FxHashMap<&'static str, Arc<dyn MessageDispatcherInterface>>>,
+}
+
+impl MessageSubsystem {
+    /// Create a new message subsystem.
+    pub fn new() -> Self {
+        MessageSubsystem { dispatchers: Mutex::new(FxHashMap::default()) }
+    }
+
+    /// Add a new message dispatcher.
+    pub async fn add_dispatch<M: Message>(&self) {
+        self.dispatchers.lock().await.insert(M::name(), Arc::new(MessageDispatcher::<M>::new()));
+    }
+
+    /// Add a dispatcher to the list of subscribers.
+    pub async fn subscribe<M: Message>(&self) -> Result<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(Error::OperationFailed)
+            }
+        };
+
+        Ok(sub)
+    }
+
+    /// Sends a message out to subscribers. Returns an error if the message
+    /// doesn't send.
+    pub async fn notify(&self, command: &str, payload: Vec<u8>) {
+        let dispatcher = self.dispatchers.lock().await.get(command).cloned();
+
+        match dispatcher {
+            Some(dispatcher) => {
+                dispatcher.trigger(payload).await;
+            }
+            None => {
+                warn!(
+                    "MessageSubsystem::notify(\"{}\", payload) did not find a dispatcher",
+                    command
+                );
+            }
+        }
+    }
+
+    /// Send a message to all subscriber channels. Clear any inactive channels.
+    pub async fn trigger_error(&self, err: Error) {
+        // TODO: this could be parallelized
+        for dispatcher in self.dispatchers.lock().await.values() {
+            dispatcher.trigger_error(err.clone()).await;
+        }
+    }
+}
+
+impl Default for MessageSubsystem {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Test functions for message subsystem.
+// This is a test function for the message subsystem code above
+// Normall we would use the #[test] macro but cannot since it is async code
+// Instead we call it using smol::block_on() in the unit test code after this
+// func
+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(Error::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/net3/mod.rs

@@ -0,0 +1,101 @@
+/// Acceptor class handles the acceptance of inbound socket connections. It's
+/// used to start listening on a local socket address, to accept incoming
+/// connections and to handle network errors.
+pub mod acceptor;
+
+/// Async channel that handles the sending of messages across the network.
+/// Public interface is used to create new channels, to stop and start
+/// a channel, and to send messages.
+///
+/// Implements message functionality and the message subscriber subsystem.
+pub mod channel;
+
+/// Handles the creation of outbound connections. Used to establish an outbound
+/// connection.
+pub mod connector;
+
+/// Hosts are a list of network addresses used when establishing an outbound
+/// connection. Hosts are shared across the network through the address
+/// protocol. When attempting to connect, a node will loop through addresses in
+/// the host store until it finds ones to connect to.
+pub mod hosts;
+
+/// Generic publish/subscribe class that can dispatch any kind of message to a
+/// subscribed list of dispatchers. Dispatchers subscribe to a single
+/// message format of any type. This is a generalized version of the simple
+/// publish-subscribe class in system::Subscriber.
+///
+/// Message Subsystem also enables the creation of new message subsystems,
+/// adding new dispatchers and clearing inactive channels.
+///
+/// Message Subsystem maintains a list of dispatchers, which is a generalized
+/// version of a subscriber. Pub-sub is called on dispatchers through the
+/// functions 'subscribe' and 'notify'. Whereas system::Subscriber only allows
+/// messages of a single type, dispatchers can handle any kind of message. This
+/// generic message is called a payload and is processed and decoded by the
+/// Message Dispatcher.
+///
+/// The Message Dispatcher is a class of subscribers that implements a
+/// generic trait called Message Dispatcher Interface, which allows us to
+/// process any kind of payload as a message.
+pub mod message_subscriber;
+
+/// Defines how to decode generic messages as well as implementing the common
+/// network messages that are sent between nodes as described by the Protocol
+/// submodule.
+///
+/// Implements a type called Packet which is the base message type. Packets are
+/// converted into messages and passed to an event loop.
+pub mod message;
+
+/// P2P provides all core functionality to interact with the peer-to-peer
+/// network.
+///
+/// Used to create a network, to start and run it, to broadcast messages across
+/// all channels, and to manage the channel store.
+///
+/// The channel store is a hashmap of channel address that we can use to add and
+/// remove channels or check whether a channel is already is in the store.
+pub mod p2p;
+
+/// Defines the networking protocol used at each stage in a connection. Consists
+/// of a series of messages that are sent across the network at the different
+/// connection stages.
+///
+/// When a node connects to a network for the first time, it must follow a seed
+/// protocol, which provides it with a list of network hosts to connect to. To
+/// establish a connection to another node, nodes must send version and version
+/// acknowledgement messages. During a connection, nodes continually get address
+/// and get-address messages to inform eachother about what nodes are on the
+/// network. Nodes also send out a ping and pong message which keeps the network
+/// from shutting down.
+///
+/// Protocol submodule also implements a jobs manager than handles the
+/// asynchronous execution of the protocols.
+pub mod protocol;
+
+/// Defines the interaction between nodes during a connection. Consists of an
+/// inbound session, which describes how to set up an incoming connection, and
+/// an outbound session, which describes setting up an outbound connection. Also
+/// describes the seed session, which is the type of connection used when a node
+/// connects to the network for the first time. Implements the session trait
+/// which describes the common functions across all sessions.
+pub mod session;
+
+/// Network configuration settings.
+pub mod settings;
+
+/// Network transport implementations.
+pub mod transport;
+
+pub use acceptor::{Acceptor, AcceptorPtr};
+pub use channel::{Channel, ChannelPtr};
+pub use connector::Connector;
+pub use hosts::{Hosts, HostsPtr};
+pub use message::Message;
+pub use message_subscriber::MessageSubscription;
+pub use p2p::{P2p, P2pPtr};
+pub use protocol::{ProtocolBase, ProtocolBasePtr, ProtocolJobsManager, ProtocolJobsManagerPtr};
+pub use session::{SESSION_ALL, SESSION_INBOUND, SESSION_MANUAL, SESSION_OUTBOUND, SESSION_SEED};
+pub use settings::{Settings, SettingsPtr};
+pub use transport::Transport;

src/net3/p2p.rs

@@ -0,0 +1,244 @@
+use async_std::sync::Mutex;
+use std::{fmt, net::SocketAddr, sync::Arc};
+
+use async_executor::Executor;
+use fxhash::{FxHashMap, FxHashSet};
+use log::debug;
+use serde_json::json;
+
+use crate::{
+    system::{Subscriber, SubscriberPtr, Subscription},
+    Error, Result,
+};
+
+use super::{
+    message::Message,
+    protocol::{register_default_protocols, ProtocolRegistry},
+    session::{InboundSession, ManualSession, OutboundSession, SeedSession, Session},
+    Channel, ChannelPtr, Hosts, HostsPtr, Settings, SettingsPtr,
+};
+
+/// List of channels that are awaiting connection.
+pub type PendingChannels = Mutex<FxHashSet<SocketAddr>>;
+/// List of connected channels.
+pub type ConnectedChannels = Mutex<fxhash::FxHashMap<SocketAddr, Arc<Channel>>>;
+/// Atomic pointer to p2p interface.
+pub type P2pPtr = Arc<P2p>;
+
+enum P2pState {
+    // The p2p object has been created but not yet started.
+    Open,
+    // We are performing the initial seed session
+    Start,
+    // Seed session finished, but not yet running
+    Started,
+    // p2p is running and the network is active.
+    Run,
+}
+
+impl fmt::Display for P2pState {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                Self::Open => "open",
+                Self::Start => "start",
+                Self::Started => "started",
+                Self::Run => "run",
+            }
+        )
+    }
+}
+
+/// Top level peer-to-peer networking interface.
+pub struct P2p {
+    pending: PendingChannels,
+    channels: ConnectedChannels,
+    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
+    // Used both internally and externally
+    stop_subscriber: SubscriberPtr<Error>,
+    hosts: HostsPtr,
+    protocol_registry: ProtocolRegistry,
+
+    // We keep a reference to the sessions used for get info
+    session_manual: Mutex<Option<Arc<ManualSession>>>,
+    session_inbound: Mutex<Option<Arc<InboundSession>>>,
+    session_outbound: Mutex<Option<Arc<OutboundSession>>>,
+
+    state: Mutex<P2pState>,
+
+    settings: SettingsPtr,
+}
+
+impl P2p {
+    /// Create a new p2p network.
+    pub async fn new(settings: Settings) -> Arc<Self> {
+        let settings = Arc::new(settings);
+
+        let self_ = Arc::new(Self {
+            pending: Mutex::new(FxHashSet::default()),
+            channels: Mutex::new(FxHashMap::default()),
+            channel_subscriber: Subscriber::new(),
+            stop_subscriber: Subscriber::new(),
+            hosts: Hosts::new(),
+            protocol_registry: ProtocolRegistry::new(),
+            session_manual: Mutex::new(None),
+            session_inbound: Mutex::new(None),
+            session_outbound: Mutex::new(None),
+            state: Mutex::new(P2pState::Open),
+            settings,
+        });
+
+        let parent = Arc::downgrade(&self_);
+
+        *self_.session_manual.lock().await = Some(ManualSession::new(parent.clone()));
+        *self_.session_inbound.lock().await = Some(InboundSession::new(parent.clone()));
+        *self_.session_outbound.lock().await = Some(OutboundSession::new(parent));
+
+        register_default_protocols(self_.clone()).await;
+
+        self_
+    }
+
+    pub async fn get_info(&self) -> serde_json::Value {
+        let external_addr = self
+            .settings
+            .external_addr
+            .map(|addr| serde_json::Value::from(addr.to_string()))
+            .unwrap_or(serde_json::Value::Null);
+
+        json!({
+            "external_addr": external_addr,
+            "session_manual": self.session_manual().await.get_info().await,
+            "session_inbound": self.session_inbound().await.get_info().await,
+            "session_outbound": self.session_outbound().await.get_info().await,
+            "state": self.state.lock().await.to_string(),
+        })
+    }
+
+    /// Invoke startup and seeding sequence. Call from constructing thread.
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "net", "P2p::start() [BEGIN]");
+
+        *self.state.lock().await = P2pState::Start;
+
+        // Start seed session
+        let seed = SeedSession::new(Arc::downgrade(&self));
+        // This will block until all seed queries have finished
+        seed.start(executor.clone()).await?;
+
+        *self.state.lock().await = P2pState::Started;
+
+        debug!(target: "net", "P2p::start() [END]");
+        Ok(())
+    }
+
+    pub async fn session_manual(&self) -> Arc<ManualSession> {
+        self.session_manual.lock().await.as_ref().unwrap().clone()
+    }
+    pub async fn session_inbound(&self) -> Arc<InboundSession> {
+        self.session_inbound.lock().await.as_ref().unwrap().clone()
+    }
+    pub async fn session_outbound(&self) -> Arc<OutboundSession> {
+        self.session_outbound.lock().await.as_ref().unwrap().clone()
+    }
+
+    /// Synchronize the blockchain and then begin long running sessions,
+    /// call after start() is invoked.
+    pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "net", "P2p::run() [BEGIN]");
+
+        *self.state.lock().await = P2pState::Run;
+
+        let manual = self.session_manual().await;
+        for peer in &self.settings.peers {
+            manual.clone().connect(peer, executor.clone()).await;
+        }
+
+        let inbound = self.session_inbound().await;
+        inbound.clone().start(executor.clone()).await?;
+
+        let outbound = self.session_outbound().await;
+        outbound.clone().start(executor.clone()).await?;
+
+        let stop_sub = self.subscribe_stop().await;
+        // Wait for stop signal
+        stop_sub.receive().await;
+
+        // Stop the sessions
+        manual.stop().await;
+        inbound.stop().await;
+        outbound.stop().await;
+
+        debug!(target: "net", "P2p::run() [END]");
+        Ok(())
+    }
+
+    /// Broadcasts a message across all channels.
+    pub async fn broadcast<M: Message + Clone>(&self, message: M) -> Result<()> {
+        for channel in self.channels.lock().await.values() {
+            channel.send(message.clone()).await?;
+        }
+        Ok(())
+    }
+
+    /// Add channel address to the list of connected channels.
+    pub async fn store(&self, channel: ChannelPtr) {
+        self.channels.lock().await.insert(channel.address(), channel.clone());
+        self.channel_subscriber.notify(Ok(channel)).await;
+    }
+
+    /// Remove a channel from the list of connected channels.
+    pub async fn remove(&self, channel: ChannelPtr) {
+        self.channels.lock().await.remove(&channel.address());
+    }
+
+    /// Check whether a channel is stored in the list of connected channels.
+    pub async fn exists(&self, addr: &SocketAddr) -> bool {
+        self.channels.lock().await.contains_key(addr)
+    }
+
+    /// Add a channel to the list of pending channels.
+    pub async fn add_pending(&self, addr: SocketAddr) -> bool {
+        self.pending.lock().await.insert(addr)
+    }
+
+    /// Remove a channel from the list of pending channels.
+    pub async fn remove_pending(&self, addr: &SocketAddr) {
+        self.pending.lock().await.remove(addr);
+    }
+
+    /// Return the number of connected channels.
+    pub async fn connections_count(&self) -> usize {
+        self.channels.lock().await.len()
+    }
+
+    /// Return an atomic pointer to the default network settings.
+    pub fn settings(&self) -> SettingsPtr {
+        self.settings.clone()
+    }
+
+    /// Return an atomic pointer to the list of hosts.
+    pub fn hosts(&self) -> HostsPtr {
+        self.hosts.clone()
+    }
+
+    pub fn protocol_registry(&self) -> &ProtocolRegistry {
+        &self.protocol_registry
+    }
+
+    /// Subscribe to a channel.
+    pub async fn subscribe_channel(&self) -> Subscription<Result<ChannelPtr>> {
+        self.channel_subscriber.clone().subscribe().await
+    }
+
+    /// Subscribe to a stop signal.
+    pub async fn subscribe_stop(&self) -> Subscription<Error> {
+        self.stop_subscriber.clone().subscribe().await
+    }
+
+    pub fn channels(&self) -> &ConnectedChannels {
+        &self.channels
+    }
+}

src/net3/protocol/mod.rs

@@ -0,0 +1,71 @@
+/// Protocol for address and get-address messages. Implements how nodes exchange
+/// connection information about other nodes on the network. Address and
+/// get-address messages are exchanged continually alongside ping-pong messages
+/// as part of a network connection.
+///
+/// Protocol starts by creating a subscription to address and get address
+/// messages. Then the protocol sends out a get address message and waits for an
+/// address message. Upon receiving an address messages, nodes add the
+/// address information to their local store.
+pub mod protocol_address;
+
+/// Manages the tasks for the network protocol. Used by other connection
+/// protocols to handle asynchronous task execution across the network. Runs all
+/// tasks that are handed to it on an executor that has stopping functionality.
+pub mod protocol_jobs_manager;
+
+/// Protocol for ping-pong keep-alive messages. Implements ping message and pong
+/// response. These messages are like the network heartbeat- they are sent
+/// continually between nodes, to ensure each node is still alive and active.
+/// Ping-pong messages ensure that the network doesn't
+/// time out.
+///
+/// Protocol starts by creating a subscription to ping and pong messages. Then
+/// it starts a loop with a timer and runs ping-pong in the task manager. It
+/// sends out a ping and waits for pong reply. Then waits for ping and replies
+/// with a pong.
+pub mod protocol_ping;
+
+/// Seed server protocol. Seed server is used when connecting to the network for
+/// the first time. Returns a list of IP addresses that nodes can connect to.
+///
+/// To start the seed protocol, we create a subscription to the address message,
+/// and send our address to the seed server. Then we send a get-address message
+/// and receive an address message. We add these addresses to our internal
+/// store.
+pub mod protocol_seed;
+
+/// Protocol for version information handshake between nodes at the start of a
+/// connection. Implements the process for exchanging version information
+/// between nodes. This is the first step when establishing a p2p connection.
+///
+/// The version protocol starts of by instantiating the protocol and creating a
+/// new subscription to version and version acknowledgement messages. Then we
+/// run the protocol. Nodes send a version message and wait for a version
+/// acknowledgement, while asynchronously waiting for version info from the
+/// other node and sending the version acknowledgement.
+pub mod protocol_version;
+
+pub mod protocol_base;
+pub mod protocol_registry;
+
+pub use protocol_address::ProtocolAddress;
+pub use protocol_jobs_manager::{ProtocolJobsManager, ProtocolJobsManagerPtr};
+pub use protocol_ping::ProtocolPing;
+pub use protocol_seed::ProtocolSeed;
+pub use protocol_version::ProtocolVersion;
+
+pub use protocol_base::{ProtocolBase, ProtocolBasePtr};
+pub use protocol_registry::ProtocolRegistry;
+
+use super::{
+    session::{SESSION_ALL, SESSION_SEED},
+    P2pPtr,
+};
+
+pub async fn register_default_protocols(p2p: P2pPtr) {
+    let registry = p2p.protocol_registry();
+    registry.register(SESSION_ALL, ProtocolPing::init).await;
+    registry.register(!SESSION_SEED, ProtocolAddress::init).await;
+    registry.register(SESSION_SEED, ProtocolSeed::init).await;
+}

src/net3/protocol/protocol_address.rs

@@ -0,0 +1,118 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use log::debug;
+use smol::Executor;
+
+use crate::Result;
+
+use super::{
+    super::{message, message_subscriber::MessageSubscription, ChannelPtr, HostsPtr, P2pPtr},
+    ProtocolBase, ProtocolBasePtr, ProtocolJobsManager, ProtocolJobsManagerPtr,
+};
+
+/// Defines address and get-address messages.
+pub struct ProtocolAddress {
+    channel: ChannelPtr,
+    addrs_sub: MessageSubscription<message::AddrsMessage>,
+    get_addrs_sub: MessageSubscription<message::GetAddrsMessage>,
+    hosts: HostsPtr,
+    jobsman: ProtocolJobsManagerPtr,
+}
+
+impl ProtocolAddress {
+    /// Create a new address protocol. Makes an address and get-address
+    /// subscription and adds them to the address protocol instance.
+    pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
+        let hosts = p2p.hosts();
+
+        // Creates a subscription to address message.
+        let addrs_sub = channel
+            .clone()
+            .subscribe_msg::<message::AddrsMessage>()
+            .await
+            .expect("Missing addrs dispatcher!");
+
+        // Creates a subscription to get-address message.
+        let get_addrs_sub = channel
+            .clone()
+            .subscribe_msg::<message::GetAddrsMessage>()
+            .await
+            .expect("Missing getaddrs dispatcher!");
+
+        Arc::new(Self {
+            channel: channel.clone(),
+            addrs_sub,
+            get_addrs_sub,
+            hosts,
+            jobsman: ProtocolJobsManager::new("ProtocolAddress", channel),
+        })
+    }
+
+    /// Handles receiving the address message. Loops to continually recieve
+    /// address messages on the address subsciption. Adds the recieved
+    /// addresses to the list of hosts.
+    async fn handle_receive_addrs(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net", "ProtocolAddress::handle_receive_addrs() [START]");
+        loop {
+            let addrs_msg = self.addrs_sub.receive().await?;
+
+            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;
+        }
+    }
+
+    /// Handles receiving the get-address message. Continually recieves
+    /// get-address messages on the get-address subsciption. Then replies
+    /// with an address message.
+    async fn handle_receive_get_addrs(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() [START]");
+        loop {
+            let _get_addrs = self.get_addrs_sub.receive().await?;
+
+            debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() received GetAddrs message");
+
+            // Loads the list of hosts.
+            let addrs = self.hosts.load_all().await;
+            debug!(
+            target: "net",
+            "ProtocolAddress::handle_receive_get_addrs() sending {} addrs",
+            addrs.len()
+            );
+            // Creates an address messages containing host address.
+            let addrs_msg = message::AddrsMessage { addrs };
+            // Sends the address message across the channel.
+            self.channel.clone().send(addrs_msg).await?;
+        }
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolAddress {
+    /// Starts the address protocol. Runs receive address and get address
+    /// protocols on the protocol task manager. Then sends get-address
+    /// message.
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "net", "ProtocolAddress::start() [START]");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().handle_receive_addrs(), executor.clone()).await;
+        self.jobsman.clone().spawn(self.clone().handle_receive_get_addrs(), executor).await;
+
+        // Send get_address message.
+        let get_addrs = message::GetAddrsMessage {};
+        let _ = self.channel.clone().send(get_addrs).await;
+        debug!(target: "net", "ProtocolAddress::start() [END]");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolAddress"
+    }
+}

src/net3/protocol/protocol_base.rs

@@ -0,0 +1,15 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use smol::Executor;
+
+use crate::Result;
+
+pub type ProtocolBasePtr = Arc<dyn ProtocolBase + Send + Sync>;
+
+#[async_trait]
+pub trait ProtocolBase {
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()>;
+
+    fn name(&self) -> &'static str;
+}

src/net3/protocol/protocol_jobs_manager.rs

@@ -0,0 +1,71 @@
+use async_std::sync::Mutex;
+use std::sync::Arc;
+
+use futures::Future;
+use log::*;
+use smol::Task;
+
+use crate::{system::ExecutorPtr, Result};
+
+use super::super::ChannelPtr;
+
+/// Pointer to protocol jobs manager.
+pub type ProtocolJobsManagerPtr = Arc<ProtocolJobsManager>;
+
+/// Manages the tasks for the network protocol. Used by other connection
+/// protocols to handle asynchronous task execution across the network. Runs all
+/// tasks that are handed to it on an executor that has stopping functionality.
+pub struct ProtocolJobsManager {
+    name: &'static str,
+    channel: ChannelPtr,
+    tasks: Mutex<Vec<Task<Result<()>>>>,
+}
+
+impl ProtocolJobsManager {
+    /// Create a new protocol jobs manager.
+    pub fn new(name: &'static str, channel: ChannelPtr) -> Arc<Self> {
+        Arc::new(Self { name, channel, tasks: Mutex::new(Vec::new()) })
+    }
+
+    /// Runs the task on an executor. Prepares to stop all tasks when the
+    /// channel is closed.
+    pub fn start(self: Arc<Self>, executor: ExecutorPtr<'_>) {
+        executor.spawn(self.handle_stop()).detach()
+    }
+
+    /// Spawns a new task and adds it to the internal queue.
+    pub async fn spawn<'a, F>(&self, future: F, executor: ExecutorPtr<'a>)
+    where
+        F: Future<Output = Result<()>> + Send + 'a,
+    {
+        self.tasks.lock().await.push(executor.spawn(future))
+    }
+
+    /// Waits for a stop signal, then closes all tasks. Insures that all tasks
+    /// are stopped when a channel closes. Called in start().
+    async fn handle_stop(self: Arc<Self>) {
+        let stop_sub = self.channel.clone().subscribe_stop().await;
+
+        // Wait for the stop signal
+        // Not interested in the exact error
+        let _ = stop_sub.receive().await;
+
+        self.close_all_tasks().await
+    }
+
+    /// Closes all open tasks. Takes all the tasks from the internal queue and
+    /// closes them.
+    async fn close_all_tasks(self: Arc<Self>) {
+        debug!(target: "net",
+            "ProtocolJobsManager::close_all_tasks() [START, name={}, addr={}]",
+            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/net3/protocol/protocol_ping.rs

@@ -0,0 +1,125 @@
+use std::{sync::Arc, time::Instant};
+
+use async_trait::async_trait;
+use log::{debug, error};
+use rand::Rng;
+use smol::Executor;
+
+use crate::{util::sleep, Error, Result};
+
+use super::{
+    super::{message, message_subscriber::MessageSubscription, ChannelPtr, P2pPtr, SettingsPtr},
+    ProtocolBase, ProtocolBasePtr, ProtocolJobsManager, ProtocolJobsManagerPtr,
+};
+
+/// Defines ping and pong messages.
+pub struct ProtocolPing {
+    channel: ChannelPtr,
+    ping_sub: MessageSubscription<message::PingMessage>,
+    pong_sub: MessageSubscription<message::PongMessage>,
+    settings: SettingsPtr,
+    jobsman: ProtocolJobsManagerPtr,
+}
+
+impl ProtocolPing {
+    /// Create a new ping-pong protocol.
+    pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
+        let settings = p2p.settings();
+
+        // Creates a subscription to ping message.
+        let ping_sub = channel
+            .clone()
+            .subscribe_msg::<message::PingMessage>()
+            .await
+            .expect("Missing ping dispatcher!");
+
+        // Creates a subscription to pong message.
+        let pong_sub = channel
+            .clone()
+            .subscribe_msg::<message::PongMessage>()
+            .await
+            .expect("Missing pong dispatcher!");
+
+        Arc::new(Self {
+            channel: channel.clone(),
+            ping_sub,
+            pong_sub,
+            settings,
+            jobsman: ProtocolJobsManager::new("ProtocolPing", channel),
+        })
+    }
+
+    /// Runs ping-pong protocol. Creates a subscription to pong, then starts a
+    /// loop. Loop sleeps for the duration of the channel heartbeat, then
+    /// sends a ping message with a random nonce. Loop starts a timer, waits
+    /// for the pong reply and insures the nonce is the same.
+    async fn run_ping_pong(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net", "ProtocolPing::run_ping_pong() [START]");
+        loop {
+            // Wait channel_heartbeat amount of time.
+            sleep(self.settings.channel_heartbeat_seconds.into()).await;
+
+            // Create a random nonce.
+            let nonce = Self::random_nonce();
+
+            // Send ping message.
+            let ping = message::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 = self.pong_sub.receive().await?;
+            if pong_msg.nonce != nonce {
+                // TODO: this is too extreme
+                error!("Wrong nonce for ping reply. Disconnecting from channel.");
+                self.channel.stop().await;
+                return Err(Error::ChannelStopped)
+            }
+            let duration = start.elapsed().as_millis();
+            debug!(target: "net", "Received Pong message {}ms from [{:?}]",
+                   duration, self.channel.address());
+        }
+    }
+
+    /// Waits for ping, then replies with pong. Copies ping's nonce into the
+    /// pong reply.
+    async fn reply_to_ping(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net", "ProtocolPing::reply_to_ping() [START]");
+        loop {
+            // Wait for ping, reply with pong that has a matching nonce.
+            let ping = self.ping_sub.receive().await?;
+            debug!(target: "net", "ProtocolPing::reply_to_ping() received Ping message");
+
+            // Send pong message.
+            let pong = message::PongMessage { nonce: ping.nonce };
+            self.channel.clone().send(pong).await?;
+            debug!(target: "net", "ProtocolPing::reply_to_ping() sent Pong reply");
+        }
+    }
+
+    fn random_nonce() -> u32 {
+        let mut rng = rand::thread_rng();
+        rng.gen()
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolPing {
+    /// Starts ping-pong keep-alive messages exchange. Runs ping-pong in the
+    /// protocol task manager, then queues the reply. Sends out a ping and
+    /// waits for pong reply. Waits for ping and replies with a pong.
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "net", "ProtocolPing::start() [START]");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman.clone().spawn(self.clone().run_ping_pong(), executor.clone()).await;
+        self.jobsman.clone().spawn(self.reply_to_ping(), executor).await;
+        debug!(target: "net", "ProtocolPing::start() [END]");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolPing"
+    }
+}

src/net3/protocol/protocol_registry.rs

@@ -0,0 +1,61 @@
+use async_std::sync::Mutex;
+use std::future::Future;
+
+use futures::future::BoxFuture;
+use log::debug;
+
+use super::{
+    super::{session::SessionBitflag, ChannelPtr, P2pPtr},
+    ProtocolBasePtr,
+};
+
+type Constructor =
+    Box<dyn Fn(ChannelPtr, P2pPtr) -> BoxFuture<'static, ProtocolBasePtr> + Send + Sync>;
+
+pub struct ProtocolRegistry {
+    protocol_constructors: Mutex<Vec<(SessionBitflag, Constructor)>>,
+}
+
+impl Default for ProtocolRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl ProtocolRegistry {
+    pub fn new() -> Self {
+        Self { protocol_constructors: Mutex::new(Vec::new()) }
+    }
+
+    // add_protocol()?
+    pub async fn register<C, F>(&self, session_flags: SessionBitflag, constructor: C)
+    where
+        C: 'static + Fn(ChannelPtr, P2pPtr) -> F + Send + Sync,
+        F: 'static + Future<Output = ProtocolBasePtr> + Send,
+    {
+        let constructor = move |channel, p2p| {
+            Box::pin(constructor(channel, p2p)) as BoxFuture<'static, ProtocolBasePtr>
+        };
+        self.protocol_constructors.lock().await.push((session_flags, Box::new(constructor)));
+    }
+
+    pub async fn attach(
+        &self,
+        selector_id: SessionBitflag,
+        channel: ChannelPtr,
+        p2p: P2pPtr,
+    ) -> Vec<ProtocolBasePtr> {
+        let mut protocols: Vec<ProtocolBasePtr> = Vec::new();
+        for (session_flags, construct) in self.protocol_constructors.lock().await.iter() {
+            // Skip protocols that are not registered for this session
+            if selector_id & session_flags == 0 {
+                continue
+            }
+
+            let protocol: ProtocolBasePtr = construct(channel.clone(), p2p.clone()).await;
+            debug!(target: "net", "Attached {}", protocol.name());
+            protocols.push(protocol)
+        }
+        protocols
+    }
+}

src/net3/protocol/protocol_seed.rs

@@ -0,0 +1,80 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use log::debug;
+use smol::Executor;
+
+use crate::Result;
+
+use super::{
+    super::{message, ChannelPtr, HostsPtr, P2pPtr, SettingsPtr},
+    ProtocolBase, ProtocolBasePtr,
+};
+
+/// Implements the seed protocol.
+pub struct ProtocolSeed {
+    channel: ChannelPtr,
+    hosts: HostsPtr,
+    settings: SettingsPtr,
+}
+
+impl ProtocolSeed {
+    /// Create a new seed protocol.
+    pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
+        let hosts = p2p.hosts();
+        let settings = p2p.settings();
+
+        Arc::new(Self { channel, hosts, settings })
+    }
+
+    /// Sends own external address over a channel. Imports own external address
+    /// from settings, then adds that address to an address message and
+    /// sends it out over the channel.
+    pub async fn send_self_address(&self) -> Result<()> {
+        match self.settings.external_addr {
+            Some(addr) => {
+                debug!(target: "net", "ProtocolSeed::send_own_address() addr={}", addr);
+                let addr = message::AddrsMessage { addrs: vec![addr] };
+                Ok(self.channel.clone().send(addr).await?)
+            }
+            // Do nothing if external address is not configured
+            None => Ok(()),
+        }
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolSeed {
+    /// Starts the seed protocol. Creates a subscription to the address message,
+    /// then sends our address to the seed server. Sends a get-address
+    /// message and receives an address message.
+    async fn start(self: Arc<Self>, _executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "net", "ProtocolSeed::start() [START]");
+        // Create a subscription to address message.
+        let addr_sub = self
+            .channel
+            .clone()
+            .subscribe_msg::<message::AddrsMessage>()
+            .await
+            .expect("Missing addrs dispatcher!");
+
+        // Send own address to the seed server.
+        self.send_self_address().await?;
+
+        // Send get address message.
+        let get_addr = message::GetAddrsMessage {};
+        self.channel.clone().send(get_addr).await?;
+
+        // Receive addresses.
+        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(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolSeed"
+    }
+}

src/net3/protocol/protocol_version.rs

@@ -0,0 +1,103 @@
+use async_std::future::timeout;
+use std::{sync::Arc, time::Duration};
+
+use log::*;
+use smol::Executor;
+
+use crate::{Error, Result};
+
+use super::super::{message, message_subscriber::MessageSubscription, ChannelPtr, SettingsPtr};
+
+/// Implements the protocol version handshake sent out by nodes at the beginning
+/// of a connection.
+pub struct ProtocolVersion {
+    channel: ChannelPtr,
+    version_sub: MessageSubscription<message::VersionMessage>,
+    verack_sub: MessageSubscription<message::VerackMessage>,
+    settings: SettingsPtr,
+}
+
+impl ProtocolVersion {
+    /// Create a new version protocol. Makes a version and version
+    /// acknowledgement subscription, then adds them to a version protocol
+    /// instance.
+    pub async fn new(channel: ChannelPtr, settings: SettingsPtr) -> Arc<Self> {
+        // Creates a version subscription.
+        let version_sub = channel
+            .clone()
+            .subscribe_msg::<message::VersionMessage>()
+            .await
+            .expect("Missing version dispatcher!");
+
+        // Creates a version acknowledgement subscription.
+        let verack_sub = channel
+            .clone()
+            .subscribe_msg::<message::VerackMessage>()
+            .await
+            .expect("Missing verack dispatcher!");
+
+        Arc::new(Self { channel, version_sub, verack_sub, settings })
+    }
+    /// Start version information exchange. Start the timer. Send version info
+    /// and wait for version acknowledgement. Wait for version info and send
+    /// version acknowledgement.
+    pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "net", "ProtocolVersion::run() [START]");
+        // Start timer
+        // Send version, wait for verack
+        // Wait for version, send verack
+        // Fin.
+        let result = match timeout(
+            Duration::from_secs(self.settings.channel_handshake_seconds.into()),
+            self.clone().exchange_versions(executor),
+        )
+        .await
+        {
+            Ok(t) => t,
+            Err(_) => Err(Error::ChannelTimeout),
+        };
+        debug!(target: "net", "ProtocolVersion::run() [END]");
+        result
+    }
+    /// Send and recieve version information.
+    async fn exchange_versions(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "net", "ProtocolVersion::exchange_versions() [START]");
+
+        let send = executor.spawn(self.clone().send_version());
+        let recv = executor.spawn(self.recv_version());
+
+        send.await?;
+        recv.await?;
+
+        debug!(target: "net", "ProtocolVersion::exchange_versions() [END]");
+        Ok(())
+    }
+    /// Send version info and wait for version acknowledgement.
+    async fn send_version(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net", "ProtocolVersion::send_version() [START]");
+        let version = message::VersionMessage {};
+        self.channel.clone().send(version).await?;
+
+        // Wait for version acknowledgement
+        let _verack_msg = self.verack_sub.receive().await?;
+
+        debug!(target: "net", "ProtocolVersion::send_version() [END]");
+        Ok(())
+    }
+    /// Recieve version info, check the message is okay and send version
+    /// acknowledgement.
+    async fn recv_version(self: Arc<Self>) -> Result<()> {
+        debug!(target: "net", "ProtocolVersion::recv_version() [START]");
+        // Rec
+        let _version_msg = self.version_sub.receive().await?;
+
+        // Check the message is OK
+
+        // Send version acknowledgement
+        let verack = message::VerackMessage {};
+        self.channel.clone().send(verack).await?;
+
+        debug!(target: "net", "ProtocolVersion::recv_version() [END]");
+        Ok(())
+    }
+}

src/net3/session/inbound_session.rs

@@ -0,0 +1,157 @@
+use async_std::sync::Mutex;
+use std::{
+    net::SocketAddr,
+    sync::{Arc, Weak},
+};
+
+use async_executor::Executor;
+use async_trait::async_trait;
+use fxhash::FxHashMap;
+use log::{error, info};
+use serde_json::json;
+
+use crate::{
+    system::{StoppableTask, StoppableTaskPtr},
+    Error, Result,
+};
+
+use super::{
+    super::{Acceptor, AcceptorPtr, ChannelPtr, P2p},
+    Session, SessionBitflag, SESSION_INBOUND,
+};
+
+struct InboundInfo {
+    channel: ChannelPtr,
+}
+
+impl InboundInfo {
+    async fn get_info(&self) -> serde_json::Value {
+        self.channel.get_info().await
+    }
+}
+
+/// Defines inbound connections session.
+pub struct InboundSession {
+    p2p: Weak<P2p>,
+    acceptor: AcceptorPtr,
+    accept_task: StoppableTaskPtr,
+    connect_infos: Mutex<FxHashMap<SocketAddr, InboundInfo>>,
+}
+
+impl InboundSession {
+    /// Create a new inbound session.
+    pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
+        let acceptor = Acceptor::new();
+
+        Arc::new(Self {
+            p2p,
+            acceptor,
+            accept_task: StoppableTask::new(),
+            connect_infos: Mutex::new(FxHashMap::default()),
+        })
+    }
+
+    /// Starts the inbound session. Begins by accepting connections and fails if
+    /// the address is not configured. Then runs the channel subscription
+    /// loop.
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        match self.p2p().settings().inbound {
+            Some(accept_addr) => {
+                self.clone().start_accept_session(accept_addr, executor.clone()).await?;
+            }
+            None => {
+                info!(target: "net", "Not configured for accepting incoming connections.");
+                return Ok(())
+            }
+        }
+
+        self.accept_task.clone().start(
+            self.clone().channel_sub_loop(executor.clone()),
+            // Ignore stop handler
+            |_| async {},
+            Error::ServiceStopped,
+            executor,
+        );
+
+        Ok(())
+    }
+    /// Stops the inbound session.
+    pub async fn stop(&self) {
+        self.acceptor.stop().await;
+        self.accept_task.stop().await;
+    }
+    /// Start accepting connections for inbound session.
+    async fn start_accept_session(
+        self: Arc<Self>,
+        accept_addr: SocketAddr,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        info!(target: "net", "Starting inbound session on {}", accept_addr);
+        let result = self.acceptor.clone().start(accept_addr, executor).await;
+        if let Err(err) = result.clone() {
+            error!(target: "net", "Error starting listener: {}", err);
+        }
+        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<'_>>) -> Result<()> {
+        let channel_sub = self.acceptor.clone().subscribe().await;
+        loop {
+            let channel = channel_sub.receive().await?;
+            // Spawn a detached task to process the channel
+            // This will just perform the channel setup then exit.
+            executor.spawn(self.clone().setup_channel(channel, executor.clone())).detach();
+        }
+    }
+
+    /// Registers the channel. First performs a network handshake and starts the
+    /// channel. Then starts sending keep-alive and address messages across the
+    /// channel.
+    async fn setup_channel(
+        self: Arc<Self>,
+        channel: ChannelPtr,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        info!(target: "net", "Connected inbound [{}]", channel.address());
+
+        self.clone().register_channel(channel.clone(), executor.clone()).await?;
+
+        self.manage_channel_for_get_info(channel).await;
+
+        Ok(())
+    }
+
+    async fn manage_channel_for_get_info(&self, channel: ChannelPtr) {
+        let key = channel.address();
+        self.connect_infos.lock().await.insert(key, InboundInfo { channel: channel.clone() });
+
+        let stop_sub = channel.subscribe_stop().await;
+        stop_sub.receive().await;
+
+        self.connect_infos.lock().await.remove(&key);
+    }
+}
+
+#[async_trait]
+impl Session for InboundSession {
+    async fn get_info(&self) -> serde_json::Value {
+        let mut infos = FxHashMap::default();
+        for (addr, info) in self.connect_infos.lock().await.iter() {
+            infos.insert(addr.to_string(), info.get_info().await);
+        }
+
+        json!({
+            "connected": infos,
+        })
+    }
+
+    fn p2p(&self) -> Arc<P2p> {
+        self.p2p.upgrade().unwrap()
+    }
+
+    fn selector_id(&self) -> SessionBitflag {
+        SESSION_INBOUND
+    }
+}

src/net3/session/manual_session.rs

@@ -0,0 +1,151 @@
+use async_std::sync::Mutex;
+use std::{
+    net::SocketAddr,
+    sync::{Arc, Weak},
+};
+
+use async_executor::Executor;
+use async_trait::async_trait;
+use log::*;
+use serde_json::json;
+
+use crate::{
+    system::{StoppableTask, StoppableTaskPtr},
+    util::sleep,
+    Error, Result,
+};
+
+use super::{
+    super::{Connector, P2p},
+    Session, SessionBitflag, SESSION_MANUAL,
+};
+
+pub struct ManualSession {
+    p2p: Weak<P2p>,
+    connect_slots: Mutex<Vec<StoppableTaskPtr>>,
+}
+
+impl ManualSession {
+    /// Create a new inbound session.
+    pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
+        Arc::new(Self { p2p, connect_slots: Mutex::new(Vec::new()) })
+    }
+
+    /// Stop the outbound session.
+    pub async fn stop(&self) {
+        let connect_slots = &*self.connect_slots.lock().await;
+
+        for slot in connect_slots {
+            slot.stop().await;
+        }
+    }
+
+    pub async fn connect(self: Arc<Self>, addr: &SocketAddr, executor: Arc<Executor<'_>>) {
+        let task = StoppableTask::new();
+
+        task.clone().start(
+            self.clone().channel_connect_loop(*addr, executor.clone()),
+            // Ignore stop handler
+            |_| async {},
+            Error::ServiceStopped,
+            executor.clone(),
+        );
+
+        self.connect_slots.lock().await.push(task);
+    }
+
+    pub async fn channel_connect_loop(
+        self: Arc<Self>,
+        addr: SocketAddr,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        let connector = Connector::new(self.p2p().settings());
+        let settings = self.p2p().settings();
+
+        let attempts = settings.manual_attempt_limit;
+        let mut remaining = attempts;
+
+        loop {
+            // Loop forever if attempts is 0
+            // Otherwise loop attempts number of times
+            remaining = if attempts == 0 { 1 } else { remaining - 1 };
+            if remaining == 0 {
+                break
+            }
+
+            self.p2p().add_pending(addr).await;
+
+            info!(target: "net", "Connecting to manual outbound [{}]", addr);
+
+            match connector.connect(addr).await {
+                Ok(channel) => {
+                    // Blacklist goes here
+
+                    info!(target: "net", "Connected to manual outbound [{}]", addr);
+
+                    let stop_sub = channel.subscribe_stop().await;
+
+                    self.clone().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?;
+
+                    // Wait for channel to close
+                    stop_sub.receive().await;
+                }
+                Err(err) => {
+                    info!(target: "net", "Unable to connect to manual outbound [{}]: {}", addr, err);
+
+                    sleep(settings.connect_timeout_seconds.into()).await;
+                }
+            }
+        }
+
+        warn!(
+        target: "net",
+        "Suspending manual connection to [{}] after {} failed attempts.",
+        addr,
+        attempts
+        );
+
+        Ok(())
+    }
+
+    // Starts sending keep-alive and address messages across the channels.
+    /*async fn attach_protocols(
+    self: Arc<Self>,
+    channel: ChannelPtr,
+    executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+    let hosts = self.p2p().hosts();
+
+    let protocol_ping = ProtocolPing::new(channel.clone(), self.p2p());
+    let protocol_addr = ProtocolAddress::new(channel, hosts).await;
+
+    protocol_ping.start(executor.clone()).await;
+    protocol_addr.start(executor).await;
+
+    Ok(())
+    }*/
+}
+
+#[async_trait]
+impl Session for ManualSession {
+    async fn get_info(&self) -> serde_json::Value {
+        json!({
+            "key": 110
+        })
+    }
+
+    fn p2p(&self) -> Arc<P2p> {
+        self.p2p.upgrade().unwrap()
+    }
+
+    fn selector_id(&self) -> SessionBitflag {
+        SESSION_MANUAL
+    }
+}

src/net3/session/mod.rs

@@ -0,0 +1,144 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use log::debug;
+use smol::Executor;
+
+use crate::Result;
+
+use super::{p2p::P2pPtr, protocol::ProtocolVersion, ChannelPtr};
+
+/// Seed connections session. Manages the creation of seed sessions. Used on
+/// first time connecting to the network. The seed node stores a list of other
+/// nodes in the network.
+pub mod seed_session;
+
+pub mod manual_session;
+
+/// Inbound connections session. Manages the creation of inbound sessions. Used
+/// to create an inbound session and start and stop the session.
+///
+/// Class consists of 3 pointers: a weak pointer to the peer-to-peer class, an
+/// acceptor pointer, and a stoppable task pointer. Using a weak pointer to P2P
+/// allows us to avoid circular dependencies.
+pub mod inbound_session;
+
+/// Outbound connections session. Manages the creation of outbound sessions.
+/// Used to create an outbound session and stop and start the session.
+///
+/// Class consists of a weak pointer to the peer-to-peer interface and a vector
+/// of outbound connection slots. Using a weak pointer to p2p allows us to avoid
+/// circular dependencies. The vector of slots is wrapped in a mutex lock. This
+/// is switched on everytime we instantiate a connection slot and insures that
+/// no other part of the program uses the slots at the same time.
+pub mod outbound_session;
+
+// bitwise selectors for the protocol_registry
+pub type SessionBitflag = u32;
+pub const SESSION_INBOUND: SessionBitflag = 0b0001;
+pub const SESSION_OUTBOUND: SessionBitflag = 0b0010;
+pub const SESSION_MANUAL: SessionBitflag = 0b0100;
+pub const SESSION_SEED: SessionBitflag = 0b1000;
+pub const SESSION_ALL: SessionBitflag = 0b1111;
+
+pub use inbound_session::InboundSession;
+pub use manual_session::ManualSession;
+pub use outbound_session::OutboundSession;
+pub use seed_session::SeedSession;
+
+/// Removes channel from the list of connected channels when a stop signal is
+/// received.
+async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
+    debug!(target: "net", "remove_sub_on_stop() [START]");
+    // Subscribe to stop events
+    let stop_sub = channel.clone().subscribe_stop().await;
+    // Wait for a stop event
+    let _ = stop_sub.receive().await;
+    debug!(target: "net",
+        "remove_sub_on_stop(): received stop event. Removing channel {}",
+        channel.address()
+    );
+    // Remove channel from p2p
+    p2p.remove(channel).await;
+    debug!(target: "net", "remove_sub_on_stop() [END]");
+}
+
+#[async_trait]
+/// Session trait.
+/// Defines methods that are used across sessions. Implements registering the
+/// channel and initializing the channel by performing a network handshake.
+pub trait Session: Sync {
+    /// Registers a new channel with the session. Performs a network handshake
+    /// and starts the channel.
+    async fn register_channel(
+        self: Arc<Self>,
+        channel: ChannelPtr,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        debug!(target: "net", "Session::register_channel() [START]");
+
+        // Protocols should all be initialized but not started
+        // We do this so that the protocols can begin receiving and buffering messages
+        // while the handshake protocol is ongoing.
+        // They are currently in sleep mode.
+        let p2p = self.p2p();
+        let protocols =
+            p2p.protocol_registry().attach(self.selector_id(), channel.clone(), p2p.clone()).await;
+
+        // Perform the handshake protocol
+        let protocol_version = ProtocolVersion::new(channel.clone(), self.p2p().settings()).await;
+        let handshake_task =
+            self.perform_handshake_protocols(protocol_version, channel.clone(), executor.clone());
+
+        // Switch on the channel
+        channel.start(executor.clone());
+
+        // Wait for handshake to finish.
+        handshake_task.await?;
+
+        // Now the channel is ready
+        debug!(target: "net", "Session handshake complete. Activating remaining protocols");
+
+        // Now start all the protocols
+        // They are responsible for managing their own lifetimes and
+        // correctly self destructing when the channel ends.
+        for protocol in protocols {
+            // Activate protocol
+            protocol.start(executor.clone()).await?;
+        }
+
+        debug!(target: "net", "Session::register_channel() [END]");
+        Ok(())
+    }
+
+    /// Performs network handshake to initialize channel. Adds the channel to
+    /// the list of connected channels, and prepares to remove the channel
+    /// when a stop signal is received.
+    async fn perform_handshake_protocols(
+        &self,
+        protocol_version: Arc<ProtocolVersion>,
+        channel: ChannelPtr,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        // Perform handshake
+        protocol_version.run(executor.clone()).await?;
+
+        // Channel is now initialized
+
+        // Add channel to p2p
+        self.p2p().store(channel.clone()).await;
+
+        // Subscribe to stop, so can remove from p2p
+        executor.spawn(remove_sub_on_stop(self.p2p(), channel)).detach();
+
+        // Channel is ready for use
+        Ok(())
+    }
+
+    async fn get_info(&self) -> serde_json::Value;
+
+    /// Returns a pointer to the p2p network interface.
+    fn p2p(&self) -> P2pPtr;
+
+    fn selector_id(&self) -> u32;
+}

src/net3/session/outbound_session.rs

@@ -0,0 +1,257 @@
+use async_std::sync::Mutex;
+use std::{
+    fmt,
+    net::SocketAddr,
+    sync::{Arc, Weak},
+};
+
+use async_executor::Executor;
+use async_trait::async_trait;
+use log::{error, info};
+use rand::seq::SliceRandom;
+use serde_json::json;
+
+use crate::{
+    system::{StoppableTask, StoppableTaskPtr},
+    Error, Result,
+};
+
+use super::{
+    super::{ChannelPtr, Connector, P2p},
+    Session, SessionBitflag, SESSION_OUTBOUND,
+};
+
+#[derive(Clone)]
+enum OutboundState {
+    Open,
+    Pending,
+    Connected,
+}
+
+impl fmt::Display for OutboundState {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                Self::Open => "open",
+                Self::Pending => "pending",
+                Self::Connected => "connected",
+            }
+        )
+    }
+}
+
+#[derive(Clone)]
+struct OutboundInfo {
+    addr: Option<SocketAddr>,
+    channel: Option<ChannelPtr>,
+    state: OutboundState,
+}
+
+impl OutboundInfo {
+    async fn get_info(&self) -> serde_json::Value {
+        let addr = match self.addr {
+            Some(addr) => serde_json::Value::String(addr.to_string()),
+            None => serde_json::Value::Null,
+        };
+
+        let channel = match &self.channel {
+            Some(channel) => channel.get_info().await,
+            None => serde_json::Value::Null,
+        };
+
+        json!({
+            "addr": addr,
+            "state": self.state.to_string(),
+            "channel": channel,
+        })
+    }
+}
+
+impl Default for OutboundInfo {
+    fn default() -> Self {
+        Self { addr: None, channel: None, state: OutboundState::Open }
+    }
+}
+
+/// Defines outbound connections session.
+pub struct OutboundSession {
+    p2p: Weak<P2p>,
+    connect_slots: Mutex<Vec<StoppableTaskPtr>>,
+    slot_info: Mutex<Vec<OutboundInfo>>,
+}
+
+impl OutboundSession {
+    /// Create a new outbound session.
+    pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
+        Arc::new(Self {
+            p2p,
+            connect_slots: Mutex::new(Vec::new()),
+            slot_info: Mutex::new(Vec::new()),
+        })
+    }
+
+    /// Start the outbound session. Runs the channel connect loop.
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        let slots_count = self.p2p().settings().outbound_connections;
+        info!(target: "net", "Starting {} outbound connection slots.", slots_count);
+        // Activate mutex lock on connection slots.
+        let mut connect_slots = self.connect_slots.lock().await;
+
+        self.slot_info.lock().await.resize(slots_count as usize, Default::default());
+
+        for i in 0..slots_count {
+            let task = StoppableTask::new();
+
+            task.clone().start(
+                self.clone().channel_connect_loop(i, executor.clone()),
+                // Ignore stop handler
+                |_| async {},
+                Error::ServiceStopped,
+                executor.clone(),
+            );
+
+            connect_slots.push(task);
+        }
+
+        Ok(())
+    }
+
+    /// Stop the outbound session.
+    pub async fn stop(&self) {
+        let connect_slots = &*self.connect_slots.lock().await;
+
+        for slot in connect_slots {
+            slot.stop().await;
+        }
+    }
+
+    /// Start making outbound connections. Creates a connector object, then
+    /// starts a connect loop. Loads a valid address then tries to connect.
+    /// Once connected, registers the channel, removes it from the list of
+    /// pending channels, and starts sending messages across the channel.
+    /// Otherwise returns a network error.
+    pub async fn channel_connect_loop(
+        self: Arc<Self>,
+        slot_number: u32,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        let connector = Connector::new(self.p2p().settings());
+
+        loop {
+            let addr = self.load_address(slot_number).await?;
+            info!(target: "net", "#{} connecting to outbound [{}]", slot_number, addr);
+            {
+                let info = &mut self.slot_info.lock().await[slot_number as usize];
+                info.addr = Some(addr);
+                info.state = OutboundState::Pending;
+            }
+
+            match connector.connect(addr).await {
+                Ok(channel) => {
+                    // Blacklist goes here
+
+                    info!(target: "net", "#{} connected to outbound [{}]", slot_number, addr);
+
+                    let stop_sub = channel.subscribe_stop().await;
+
+                    self.clone().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;
+                    {
+                        let info = &mut self.slot_info.lock().await[slot_number as usize];
+                        info.channel = Some(channel.clone());
+                        info.state = OutboundState::Connected;
+                    }
+
+                    // Wait for channel to close
+                    stop_sub.receive().await;
+                }
+                Err(err) => {
+                    info!(target: "net", "Unable to connect to outbound [{}]: {}", addr, err);
+                    {
+                        let info = &mut self.slot_info.lock().await[slot_number as usize];
+                        info.addr = None;
+                        info.channel = None;
+                        info.state = OutboundState::Open;
+                    }
+                }
+            }
+        }
+    }
+
+    /// Loops through host addresses to find a outbound address that we can
+    /// connect to. Checks whether address is valid by making sure it isn't
+    /// our own inbound address, then checks whether it is already connected
+    /// (exists) or connecting (pending). Keeps looping until address is
+    /// found that passes all checks.
+    async fn load_address(&self, slot_number: u32) -> Result<SocketAddr> {
+        let p2p = self.p2p();
+        let self_inbound_addr = p2p.settings().external_addr;
+
+        let mut addrs;
+
+        {
+            let hosts = p2p.hosts().load_all().await;
+            addrs = hosts;
+        }
+
+        addrs.shuffle(&mut rand::thread_rng());
+
+        for addr in addrs {
+            if p2p.exists(&addr).await {
+                continue
+            }
+
+            // Obtain a lock on this address to prevent duplicate connections
+            if !p2p.add_pending(addr).await {
+                continue
+            }
+
+            if Self::is_self_inbound(&addr, &self_inbound_addr) {
+                continue
+            }
+
+            return Ok(addr)
+        }
+
+        error!(target: "net", "Hosts address pool is empty. Closing connect slot #{}", slot_number);
+        Err(Error::ServiceStopped)
+    }
+
+    /// Checks whether an address is our own inbound address to avoid connecting
+    /// to ourselves.
+    fn is_self_inbound(addr: &SocketAddr, inbound_addr: &Option<SocketAddr>) -> bool {
+        match inbound_addr {
+            Some(inbound_addr) => inbound_addr == addr,
+            // No inbound listening address configured
+            None => false,
+        }
+    }
+}
+
+#[async_trait]
+impl Session for OutboundSession {
+    async fn get_info(&self) -> serde_json::Value {
+        let mut slots = Vec::new();
+        for info in &*self.slot_info.lock().await {
+            slots.push(info.get_info().await);
+        }
+
+        json!({
+            "slots": slots,
+        })
+    }
+
+    fn p2p(&self) -> Arc<P2p> {
+        self.p2p.upgrade().unwrap()
+    }
+
+    fn selector_id(&self) -> SessionBitflag {
+        SESSION_OUTBOUND
+    }
+}

src/net3/session/seed_session.rs

@@ -0,0 +1,152 @@
+use async_std::future::timeout;
+use std::{
+    net::SocketAddr,
+    sync::{Arc, Weak},
+    time::Duration,
+};
+
+use async_executor::Executor;
+use async_trait::async_trait;
+use log::*;
+use serde_json::json;
+
+use crate::{Error, Result};
+
+use super::{
+    super::{Connector, P2p},
+    Session, SessionBitflag, SESSION_SEED,
+};
+
+/// Defines seed connections session.
+pub struct SeedSession {
+    p2p: Weak<P2p>,
+}
+
+impl SeedSession {
+    /// Create a new seed session instance.
+    pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
+        Arc::new(Self { p2p })
+    }
+
+    /// Start the seed session. Creates a new task for every seed connection and
+    /// starts the seed on each task.
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!(target: "net", "SeedSession::start() [START]");
+        let settings = self.p2p().settings();
+
+        if settings.seeds.is_empty() {
+            warn!("Skipping seed sync process since no seeds are configured.");
+            return Ok(())
+        }
+
+        // if cached addresses then quit
+
+        let mut tasks = Vec::new();
+
+        for (i, seed) in settings.seeds.iter().enumerate() {
+            tasks.push(executor.spawn(self.clone().start_seed(i, *seed, executor.clone())));
+        }
+
+        // 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.
+        let result =
+            timeout(Duration::from_secs(settings.seed_query_timeout_seconds.into()), 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),
+                    }
+                }
+            })
+            .await;
+
+        if result.is_err() {
+            error!("Querying seeds timed out");
+            return Err(Error::OperationFailed)
+        }
+
+        // Seed process complete
+        if self.p2p().hosts().is_empty().await {
+            error!("Hosts pool still empty after seeding");
+            return Err(Error::OperationFailed)
+        }
+
+        debug!(target: "net", "SeedSession::start() [END]");
+        Ok(())
+    }
+
+    /// Connects to a seed socket address. Registers a new channel with a
+    /// network handshake, then starts the keep-alive messages and seed
+    /// protocol.
+    async fn start_seed(
+        self: Arc<Self>,
+        seed_index: usize,
+        seed: SocketAddr,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        debug!(target: "net", "SeedSession::start_seed(i={}) [START]", seed_index);
+        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_index, seed);
+
+                self.clone().register_channel(channel.clone(), executor.clone()).await?;
+
+                //self.attach_protocols(channel, hosts, settings, executor).await?;
+
+                debug!(target: "net", "SeedSession::start_seed(i={}) [END]", seed_index);
+                Ok(())
+            }
+            Err(err) => {
+                info!("Failure contacting seed #{} [{}]: {}", seed_index, seed, err);
+                Err(err)
+            }
+        }
+    }
+
+    // Starts keep-alive messages and seed protocol.
+    /*async fn attach_protocols(
+      self: Arc<Self>,
+      channel: ChannelPtr,
+      hosts: HostsPtr,
+      settings: SettingsPtr,
+      executor: Arc<Executor<'_>>,
+      ) -> Result<()> {
+      let protocol_ping = ProtocolPing::new(channel.clone(), self.p2p());
+      protocol_ping.start(executor.clone()).await;
+
+      let protocol_seed = ProtocolSeed::new(channel.clone(), hosts, settings.clone());
+    // This will block until seed process is complete
+    protocol_seed.start(executor.clone()).await?;
+
+    channel.stop().await;
+
+    Ok(())
+    }*/
+}
+
+#[async_trait]
+impl Session for SeedSession {
+    async fn get_info(&self) -> serde_json::Value {
+        json!({
+            "key": 110
+        })
+    }
+
+    fn p2p(&self) -> Arc<P2p> {
+        self.p2p.upgrade().unwrap()
+    }
+
+    fn selector_id(&self) -> SessionBitflag {
+        SESSION_SEED
+    }
+}

src/net3/settings.rs

@@ -0,0 +1,38 @@
+use std::{net::SocketAddr, sync::Arc};
+
+/// Atomic pointer to network settings.
+pub type SettingsPtr = Arc<Settings>;
+
+/// Defines the network settings.
+#[derive(Clone)]
+pub struct Settings {
+    pub inbound: Option<SocketAddr>,
+    pub outbound_connections: u32,
+    pub manual_attempt_limit: u32,
+
+    pub seed_query_timeout_seconds: 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>,
+}
+
+impl Default for Settings {
+    fn default() -> Self {
+        Self {
+            inbound: None,
+            outbound_connections: 0,
+            manual_attempt_limit: 0,
+            seed_query_timeout_seconds: 8,
+            connect_timeout_seconds: 10,
+            channel_handshake_seconds: 4,
+            channel_heartbeat_seconds: 10,
+            external_addr: None,
+            peers: Vec::new(),
+            seeds: Vec::new(),
+        }
+    }
+}

src/net3/transport.rs

@@ -0,0 +1,47 @@
+use std::error::Error;
+
+use async_trait::async_trait;
+use futures::{
+    io::{ReadHalf, WriteHalf},
+    prelude::*,
+};
+use url::Url;
+
+mod tcp;
+mod tls;
+mod tor;
+
+pub use tcp::TcpTransport;
+pub use tls::TlsTransport;
+pub use tor::TorTransport;
+
+#[async_trait]
+pub trait Transport {
+    type Acceptor;
+    type Connector;
+
+    type Error: Error;
+
+    type Listener: Future<Output = Result<Self::Acceptor, Self::Error>>;
+    type Dial: Future<Output = Result<Self::Connector, Self::Error>>;
+
+    fn listen_on(self, url: Url) -> Result<Self::Listener, TransportError<Self::Error>>
+    where
+        Self: Sized;
+
+    fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>>
+    where
+        Self: Sized;
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TransportError<TErr> {
+    #[error("Address not supported: {0}")]
+    AddrNotSupported(Url),
+
+    #[error("Transport IO Error: {0}")]
+    IoError(#[from] std::io::Error),
+
+    #[error("{0}")]
+    Other(TErr),
+}

src/net3/transport/tcp.rs

@@ -0,0 +1,97 @@
+use async_std::{
+    net::{TcpListener, TcpStream},
+    sync::{Arc, Mutex},
+};
+use std::{io, net::SocketAddr, pin::Pin};
+
+use futures::{
+    io::{ReadHalf, WriteHalf},
+    prelude::*,
+};
+use log::debug;
+use socket2::{Domain, Socket, Type};
+use url::Url;
+
+use super::{Transport, TransportError};
+
+#[derive(Clone)]
+pub struct TcpTransport {
+    /// TTL to set for opened sockets, or `None` for default
+    ttl: Option<u32>,
+    /// Size of the listen backlog for listen sockets
+    backlog: i32,
+}
+
+impl Transport for TcpTransport {
+    type Acceptor = TcpListener;
+    type Connector = TcpStream;
+
+    type Error = io::Error;
+
+    type Listener = Pin<Box<dyn Future<Output = Result<Self::Acceptor, Self::Error>> + Send>>;
+    type Dial = Pin<Box<dyn Future<Output = Result<Self::Connector, Self::Error>> + Send>>;
+
+    fn listen_on(self, url: Url) -> Result<Self::Listener, TransportError<Self::Error>> {
+        if url.scheme() != "tcp" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        debug!(target: "tcptransport", "listening on {}", socket_addr);
+        Ok(Box::pin(self.do_listen(socket_addr)))
+    }
+
+    fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>> {
+        if url.scheme() != "tcp" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        debug!(target: "tcptransport", "dialing {}", socket_addr);
+        Ok(Box::pin(self.do_dial(socket_addr)))
+    }
+}
+
+impl TcpTransport {
+    pub fn new(ttl: Option<u32>, backlog: i32) -> Self {
+        Self { ttl, backlog }
+    }
+
+    fn create_socket(&self, socket_addr: SocketAddr) -> io::Result<Socket> {
+        let domain = if socket_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
+        let socket = Socket::new(domain, Type::STREAM, Some(socket2::Protocol::TCP))?;
+
+        if socket_addr.is_ipv6() {
+            socket.set_only_v6(true)?;
+        }
+
+        if let Some(ttl) = self.ttl {
+            socket.set_ttl(ttl)?;
+        }
+
+        Ok(socket)
+    }
+
+    async fn do_listen(self, socket_addr: SocketAddr) -> Result<TcpListener, io::Error> {
+        let socket = self.create_socket(socket_addr)?;
+        socket.bind(&socket_addr.into())?;
+        socket.listen(self.backlog)?;
+        socket.set_nonblocking(true)?;
+        Ok(TcpListener::from(std::net::TcpListener::from(socket)))
+    }
+
+    async fn do_dial(self, socket_addr: SocketAddr) -> Result<TcpStream, io::Error> {
+        let socket = self.create_socket(socket_addr)?;
+        socket.set_nonblocking(true)?;
+
+        match socket.connect(&socket_addr.into()) {
+            Ok(()) => {}
+            Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) => {}
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            Err(err) => return Err(err),
+        };
+
+        let stream = TcpStream::from(std::net::TcpStream::from(socket));
+        Ok(stream)
+    }
+}

src/net3/transport/tls.rs

@@ -0,0 +1,199 @@
+use std::{io, net::SocketAddr, pin::Pin, sync::Arc, time::SystemTime};
+
+use async_std::net::{TcpListener, TcpStream};
+use futures::prelude::*;
+use futures_rustls::{
+    rustls,
+    rustls::{
+        client::{ServerCertVerified, ServerCertVerifier},
+        kx_group::X25519,
+        server::{ClientCertVerified, ClientCertVerifier},
+        version::TLS13,
+        Certificate, ClientConfig, DistinguishedNames, ServerConfig, ServerName,
+    },
+    TlsAcceptor, TlsConnector, TlsStream,
+};
+use log::debug;
+use rustls_pemfile::pkcs8_private_keys;
+use socket2::{Domain, Socket, Type};
+use url::Url;
+
+use super::{Transport, TransportError};
+
+const CIPHER_SUITE: &str = "TLS13_CHACHA20_POLY1305_SHA256";
+
+fn cipher_suite() -> rustls::SupportedCipherSuite {
+    for suite in rustls::ALL_CIPHER_SUITES {
+        let sname = format!("{:?}", suite.suite()).to_lowercase();
+
+        if sname == CIPHER_SUITE.to_string().to_lowercase() {
+            return *suite
+        }
+    }
+
+    unreachable!()
+}
+
+struct ServerCertificateVerifier;
+impl ServerCertVerifier for ServerCertificateVerifier {
+    fn verify_server_cert(
+        &self,
+        _end_entity: &Certificate,
+        _intermediates: &[Certificate],
+        _server_name: &ServerName,
+        _scts: &mut dyn Iterator<Item = &[u8]>,
+        _ocsp_response: &[u8],
+        _now: SystemTime,
+    ) -> Result<ServerCertVerified, rustls::Error> {
+        // TODO: upsycle
+        Ok(ServerCertVerified::assertion())
+    }
+}
+
+struct ClientCertificateVerifier;
+impl ClientCertVerifier for ClientCertificateVerifier {
+    fn client_auth_root_subjects(&self) -> Option<DistinguishedNames> {
+        Some(vec![])
+    }
+
+    fn verify_client_cert(
+        &self,
+        _end_entity: &Certificate,
+        _intermediates: &[Certificate],
+        _now: SystemTime,
+    ) -> Result<ClientCertVerified, rustls::Error> {
+        // TODO: upsycle
+        Ok(ClientCertVerified::assertion())
+    }
+}
+
+#[derive(Clone)]
+pub struct TlsTransport {
+    /// TTL to set for opened sockets, or `None` for default
+    ttl: Option<u32>,
+    /// Size of the listen backlog for listen sockets
+    backlog: i32,
+    /// TLS server configuration
+    server_config: Arc<ServerConfig>,
+    /// TLS client configuration
+    client_config: Arc<ClientConfig>,
+}
+
+impl Transport for TlsTransport {
+    type Acceptor = (TlsAcceptor, TcpListener);
+    type Connector = TlsStream<TcpStream>;
+
+    type Error = io::Error;
+
+    type Listener = Pin<Box<dyn Future<Output = Result<Self::Acceptor, Self::Error>> + Send>>;
+    type Dial = Pin<Box<dyn Future<Output = Result<Self::Connector, Self::Error>> + Send>>;
+
+    fn listen_on(self, url: Url) -> Result<Self::Listener, TransportError<Self::Error>> {
+        if url.scheme() != "tls" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+
+        debug!(target: "tlstransport", "listening on {}", url);
+        Ok(Box::pin(self.do_listen(url)))
+    }
+
+    fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>> {
+        if url.scheme() != "tls" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+
+        debug!(target: "tlstransport", "dialing {}", url);
+        Ok(Box::pin(self.do_dial(url)))
+    }
+}
+
+impl TlsTransport {
+    pub fn new(ttl: Option<u32>, backlog: i32) -> Self {
+        // On each instantiation, generate a new keypair and certificate
+        let keypair_pem = ed25519_compact::KeyPair::generate().to_pem();
+        let secret_key = pkcs8_private_keys(&mut keypair_pem.as_bytes()).unwrap();
+        let secret_key = rustls::PrivateKey(secret_key[0].clone());
+
+        let altnames = vec![String::from("dark.fi")];
+        let mut cert_params = rcgen::CertificateParams::new(altnames);
+        cert_params.alg = &rcgen::PKCS_ED25519;
+        cert_params.key_pair = Some(rcgen::KeyPair::from_pem(&keypair_pem).unwrap());
+
+        let certificate = rcgen::Certificate::from_params(cert_params).unwrap();
+        let certificate = certificate.serialize_der().unwrap();
+        let certificate = rustls::Certificate(certificate);
+
+        let client_cert_verifier = Arc::new(ClientCertificateVerifier {});
+        let server_config = Arc::new(
+            ServerConfig::builder()
+                .with_cipher_suites(&[cipher_suite()])
+                .with_kx_groups(&[&X25519])
+                .with_protocol_versions(&[&TLS13])
+                .unwrap()
+                .with_client_cert_verifier(client_cert_verifier)
+                .with_single_cert(vec![certificate.clone()], secret_key.clone())
+                .unwrap(),
+        );
+
+        let server_cert_verifier = Arc::new(ServerCertificateVerifier {});
+        let client_config = Arc::new(
+            ClientConfig::builder()
+                .with_cipher_suites(&[cipher_suite()])
+                .with_kx_groups(&[&X25519])
+                .with_protocol_versions(&[&TLS13])
+                .unwrap()
+                .with_custom_certificate_verifier(server_cert_verifier)
+                .with_single_cert(vec![certificate], secret_key)
+                .unwrap(),
+        );
+
+        Self { ttl, backlog, server_config, client_config }
+    }
+
+    fn create_socket(&self, socket_addr: SocketAddr) -> io::Result<Socket> {
+        let domain = if socket_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
+        let socket = Socket::new(domain, Type::STREAM, Some(socket2::Protocol::TCP))?;
+
+        if socket_addr.is_ipv6() {
+            socket.set_only_v6(true)?;
+        }
+
+        if let Some(ttl) = self.ttl {
+            socket.set_ttl(ttl)?;
+        }
+
+        Ok(socket)
+    }
+
+    async fn do_listen(self, url: Url) -> Result<(TlsAcceptor, TcpListener), io::Error> {
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        let socket = self.create_socket(socket_addr)?;
+        socket.bind(&socket_addr.into())?;
+        socket.listen(self.backlog)?;
+        socket.set_nonblocking(true)?;
+
+        let listener = TcpListener::from(std::net::TcpListener::from(socket));
+        let acceptor = TlsAcceptor::from(self.server_config);
+        Ok((acceptor, listener))
+    }
+
+    async fn do_dial(self, url: Url) -> Result<TlsStream<TcpStream>, io::Error> {
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        let server_name = ServerName::try_from("dark.fi").unwrap();
+        let socket = self.create_socket(socket_addr)?;
+        socket.set_nonblocking(true)?;
+
+        let connector = TlsConnector::from(self.client_config);
+
+        match socket.connect(&socket_addr.into()) {
+            Ok(()) => {}
+            Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) => {}
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            Err(err) => return Err(err),
+        };
+
+        let stream = TcpStream::from(std::net::TcpStream::from(socket));
+        let stream = connector.connect(server_name, stream).await?;
+        Ok(TlsStream::Client(stream))
+    }
+}

src/net3/transport/tor.rs

@@ -0,0 +1,254 @@
+use super::{Transport, TransportError};
+use async_std::{
+    net::{TcpListener, TcpStream},
+    sync::Arc,
+};
+use fast_socks5::{
+    client::{Config, Socks5Stream},
+    Result, SocksError,
+};
+use futures::prelude::*;
+
+use regex::Regex;
+use socket2::{Domain, Socket, Type};
+use std::{
+    io,
+    io::{BufRead, BufReader, Write},
+    net::SocketAddr,
+    pin::Pin,
+    time::Duration,
+};
+use url::Url;
+
+/// Implements communication through the tor proxy service.
+///
+/// ## Dialing
+///
+/// The tor service must be running for dialing to work. Url of it has to be passed to the
+/// constructor.
+///
+/// ## Listening
+///
+/// Two ways of setting up hidden services are allowed: hidden services manually set up by the user
+/// in the torc file or ephemereal hidden services created and deleted on the fly. For the latter,
+/// the user must set up the tor control port[^controlport].
+///
+/// Having manually configured services forces the program to use pre-defined ports, i.e. it has no
+/// way of changing them.
+///
+/// Before calling [listen_on][transportlisten] on a local address, make sure that either a hidden
+/// service pointing to that address was configured or that [create_ehs][torcreateehs] was called
+/// with this address.
+///
+/// [^controlport] [Open control port](https://wiki.archlinux.org/title/tor#Open_Tor_ControlPort)
+///
+/// ### Warning on cloning
+/// Cloning this structure increments the reference count to the already open
+/// socket, which means ephemereal hidden services opened with the cloned instance will live as
+/// long as there are clones. For this reason, I'd clone it only when you are sure you want this
+/// behaviour. Don't be lazy!
+///
+/// [transportlisten]: Transport
+/// [torcreateehs]: TorTransport::create_ehs
+#[derive(Clone)]
+pub struct TorTransport {
+    socks_url: Url,
+    tor_controller: Option<TorController>,
+}
+
+/// Represents information needed to communicate with the Tor control socket
+#[derive(Clone)]
+struct TorController {
+    socket: Arc<Socket>, // Need to hold this socket open as long as the tor trasport is alive, so ephemeral services are dropped when TorTransport is dropped
+    auth: String,
+}
+
+/// Wraps the errors, because dialing and listening use different communication
+#[derive(Debug, thiserror::Error)]
+pub enum TorError {
+    #[error("Transport IO Error: {0}")]
+    IoError(#[from] io::Error),
+    #[error("Socks: {0}")]
+    Socks5Error(#[from] SocksError),
+    #[error("Url parse error: {0}")]
+    UrlParseError(#[from] url::ParseError),
+    #[error("Regex parse error: {0}")]
+    RegexError(#[from] regex::Error),
+    #[error("Unexpected response from tor: {0}")]
+    GeneralError(String),
+}
+
+/// Contains the configuration to communicate with the Tor Controler
+///
+/// When cloned, the socket is not reopened since we use reference count.
+/// The hidden services created live as long as clones of the struct.
+impl TorController {
+    /// Creates a new TorTransport
+    ///
+    /// # Arguments
+    ///
+    /// * `url` - url to connect to the tor control. For example tcp://127.0.0.1:9051
+    ///
+    /// * `auth` - either authentication cookie bytes (32 bytes) as hex in a string
+    /// or a password as a quoted string.
+    ///
+    /// Cookie string: `assert_eq!(auth,"886b9177aec471965abd34b6a846dc32cf617dcff0625cba7a414e31dd4b75a0")`
+    ///
+    /// Password string: `assert_eq!(auth,"\"mypassword\"")`
+    pub fn new(url: Url, auth: String) -> Result<Self, io::Error> {
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        let domain = if socket_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
+        let socket = Socket::new(domain, Type::STREAM, Some(socket2::Protocol::TCP))?;
+        if socket_addr.is_ipv6() {
+            socket.set_only_v6(true)?;
+        }
+
+        match socket.connect(&socket_addr.into()) {
+            Ok(()) => {}
+            Err(err) if err.raw_os_error() == Some(libc::EINPROGRESS) => {}
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {}
+            Err(err) => return Err(err),
+        };
+        Ok(Self { socket: Arc::new(socket), auth })
+    }
+    /// Creates an ephemeral hidden service pointing to local address, returns onion address
+    ///
+    /// # Arguments
+    ///
+    /// * `url` - url that the hidden service maps to.
+    pub fn create_ehs(&self, url: Url) -> Result<Url, TorError> {
+        let local_socket = self.socket.try_clone()?;
+        let mut stream = std::net::TcpStream::from(local_socket);
+
+        stream.set_write_timeout(Some(Duration::from_secs(2)))?;
+        let host = url
+            .host()
+            .ok_or_else(|| TorError::GeneralError("No host on url for listening".to_string()))?;
+        let port = url
+            .port()
+            .ok_or_else(|| TorError::GeneralError("No port on url for listening".to_string()))?;
+
+        let payload = format!(
+            "AUTHENTICATE {a}\r\nADD_ONION NEW:BEST Flags=DiscardPK Port={p},{h}:{p}\r\n",
+            a = self.auth,
+            p = port,
+            h = host
+        );
+        stream.write_all(payload.as_bytes())?;
+        stream.set_read_timeout(Some(Duration::from_secs(1)))?; // Maybe a bit too much. Gives tor time to reply
+        let mut reader = BufReader::new(stream);
+        let mut repl = String::new();
+        while let Ok(nbytes) = reader.read_line(&mut repl) {
+            if nbytes == 0 {
+                break
+            }
+        }
+        let re = Regex::new(r"250-ServiceID=(\w+*)")?;
+        let cap: Result<regex::Captures<'_>, TorError> =
+            re.captures(&repl).ok_or_else(|| TorError::GeneralError(repl.clone()));
+        let hurl =
+            cap?.get(1).map_or(Err(TorError::GeneralError(repl.clone())), |m| Ok(m.as_str()))?;
+        let hurl = format!("tcp://{}.onion:{}", &hurl, port);
+        Ok(Url::parse(&hurl)?)
+    }
+}
+
+impl TorTransport {
+    /// Creates a new TorTransport
+    ///
+    /// # Arguments
+    ///
+    /// * `socks_url` - url to connect to the tor service. For example socks5://127.0.0.1:9050
+    ///
+    /// * `control_info` - Possibility to open a control connection to create ephemeral hidden
+    /// services that live as long as the TorTransport.
+    /// It is a tuple of the control socket url and authentication cookie as string
+    /// represented in hex.
+    pub fn new(socks_url: Url, control_info: Option<(Url, String)>) -> Result<Self, TorError> {
+        match control_info {
+            Some(info) => {
+                let (url, auth) = info;
+                let tor_controller = Some(TorController::new(url, auth)?);
+                Ok(Self { socks_url, tor_controller })
+            }
+            None => Ok(Self { socks_url, tor_controller: None }),
+        }
+    }
+
+    /// Creates an ephemeral hidden service pointing to local address, returns onion address
+    /// when successful.
+    ///
+    /// # Arguments
+    ///
+    /// * `url` - url that the hidden service maps to.
+    pub fn create_ehs(&self, url: Url) -> Result<Url, TorError> {
+        self.tor_controller
+            .as_ref()
+            .ok_or_else(|| {
+                TorError::GeneralError("No controller configured for this transport".to_string())
+            })?
+            .create_ehs(url)
+    }
+
+    pub async fn do_dial(self, url: Url) -> Result<Socks5Stream<TcpStream>, TorError> {
+        let socks_url_str = self.socks_url.socket_addrs(|| None)?[0].to_string();
+        let host = url.host().unwrap().to_string();
+        let port = url.port().unwrap_or(80);
+        let config = Config::default();
+        let stream = if !self.socks_url.username().is_empty() && self.socks_url.password().is_some()
+        {
+            Socks5Stream::connect_with_password(
+                socks_url_str,
+                host,
+                port,
+                self.socks_url.username().to_string(),
+                self.socks_url.password().unwrap().to_string(),
+                config,
+            )
+            .await?
+        } else {
+            Socks5Stream::connect(socks_url_str, host, port, config).await?
+        };
+        Ok(stream)
+    }
+
+    fn create_socket(&self, socket_addr: SocketAddr) -> io::Result<Socket> {
+        let domain = if socket_addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
+        let socket = Socket::new(domain, Type::STREAM, Some(socket2::Protocol::TCP))?;
+
+        if socket_addr.is_ipv6() {
+            socket.set_only_v6(true)?;
+        }
+        Ok(socket)
+    }
+
+    pub async fn do_listen(self, url: Url) -> Result<TcpListener, TorError> {
+        let socket_addr = url.socket_addrs(|| None)?[0];
+        let socket = self.create_socket(socket_addr)?;
+        socket.bind(&socket_addr.into())?;
+        socket.listen(1024)?;
+        socket.set_nonblocking(true)?;
+        Ok(TcpListener::from(std::net::TcpListener::from(socket)))
+    }
+}
+
+impl Transport for TorTransport {
+    type Acceptor = TcpListener;
+    type Connector = Socks5Stream<TcpStream>;
+
+    type Error = TorError;
+
+    type Listener = Pin<Box<dyn Future<Output = Result<Self::Acceptor, Self::Error>> + Send>>;
+    type Dial = Pin<Box<dyn Future<Output = Result<Self::Connector, Self::Error>> + Send>>;
+
+    fn listen_on(self, url: Url) -> Result<Self::Listener, TransportError<Self::Error>> {
+        if url.scheme() != "tcp" {
+            return Err(TransportError::AddrNotSupported(url))
+        }
+        Ok(Box::pin(self.do_listen(url)))
+    }
+
+    fn dial(self, url: Url) -> Result<Self::Dial, TransportError<Self::Error>> {
+        Ok(Box::pin(self.do_dial(url)))
+    }
+}