remove net2

Cargo.toml

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

src/error.rs

@@ -344,14 +344,6 @@ impl From<VerifyFailed> for ClientFailed {
     }
 }
 
-// TEMP
-#[cfg(feature = "net2")]
-impl<T: std::fmt::Display> From<crate::net2::transport::TransportError<T>> for Error {
-    fn from(err: crate::net2::transport::TransportError<T>) -> Self {
-        Self::TransportError(err.to_string())
-    }
-}
-
 // TEMP
 #[cfg(feature = "net3")]
 impl<T: std::fmt::Display> From<crate::net3::transport::TransportError<T>> for Error {

src/lib.rs

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

src/net2/acceptor.rs

@@ -1,84 +0,0 @@
-use std::sync::Arc;
-
-use smol::Executor;
-use url::Url;
-
-use crate::{
-    system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
-    Error, Result,
-};
-
-use super::{Channel, ChannelPtr, Transport};
-
-/// Atomic pointer to Acceptor class.
-pub type AcceptorPtr<T> = Arc<Acceptor<T>>;
-
-/// Create inbound socket connections.
-pub struct Acceptor<T: Transport> {
-    channel_subscriber: SubscriberPtr<Result<ChannelPtr<T>>>,
-    task: StoppableTaskPtr,
-}
-
-impl<T: Transport> Acceptor<T> {
-    /// 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: Url,
-        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<T>>> {
-        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>, url: Url, executor: Arc<Executor<'_>>) {
-        self.task.clone().start(
-            self.clone().run_accept_loop(url),
-            |result| self.handle_stop(result),
-            Error::ServiceStopped,
-            executor,
-        );
-    }
-
-    /// Run the accept loop.
-    async fn run_accept_loop(self: Arc<Self>, url: url::Url) -> Result<()> {
-        let transport = T::new(None, 1024);
-        let listener = Arc::new(transport.listen_on(url.clone())?.await?);
-        loop {
-            let stream = T::accept(listener.clone()).await?;
-            let channel = Channel::<T>::new(stream, url.clone()).await;
-            self.channel_subscriber.notify(Ok(channel)).await;
-        }
-    }
-
-    /// 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;
-            }
-        }
-    }
-}

src/net2/channel.rs

@@ -1,289 +0,0 @@
-use async_std::sync::Mutex;
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc,
-};
-
-use futures::{
-    io::{ReadHalf, WriteHalf},
-    AsyncReadExt,
-};
-use log::{debug, error, info};
-use serde_json::json;
-use smol::Executor;
-use url::Url;
-
-use crate::{
-    system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
-    Error, Result,
-};
-
-use super::{
-    message,
-    message_subscriber::{MessageSubscription, MessageSubsystem},
-    Transport,
-};
-
-/// Atomic pointer to async channel.
-pub type ChannelPtr<T> = Arc<Channel<T>>;
-
-struct ChannelInfo {
-    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 { last_msg: String::new(), last_status: String::new(), log: Mutex::new(Vec::new()) }
-    }
-
-    async fn get_info(&self) -> serde_json::Value {
-        let result = json!({
-            "last_msg": self.last_msg,
-            "last_status": self.last_status,
-            "log": self.log.lock().await.clone(),
-        });
-        self.log.lock().await.clear();
-        result
-    }
-}
-
-/// Async channel for communication between nodes.
-pub struct Channel<T: Transport> {
-    reader: Mutex<ReadHalf<T::Connector>>,
-    writer: Mutex<WriteHalf<T::Connector>>,
-    address: Url,
-    message_subsystem: MessageSubsystem,
-    stop_subscriber: SubscriberPtr<Error>,
-    receive_task: StoppableTaskPtr,
-    stopped: AtomicBool,
-    info: Mutex<ChannelInfo>,
-}
-
-impl<T: Transport> Channel<T> {
-    /// 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: T::Connector, address: Url) -> 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) -> Url {
-        self.address.clone()
-    }
-
-    /// 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/net2/connector.rs

@@ -1,37 +0,0 @@
-use async_std::future::timeout;
-use std::time::Duration;
-
-use url::Url;
-
-use crate::{Error, Result};
-
-use super::{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<T: Transport>(&self, hostaddr: Url) -> Result<ChannelPtr<T>> {
-        let stream_result =
-            timeout(Duration::from_secs(self.settings.connect_timeout_seconds.into()), async {
-                let transport = T::new(None, 1024);
-                let connect_stream = transport.dial(hostaddr.clone())?.await?;
-                let channel = Channel::<T>::new(connect_stream, hostaddr).await;
-                Ok(channel)
-            })
-            .await;
-
-        match stream_result {
-            Ok(t) => t,
-            Err(_) => Err(Error::ConnectTimeout),
-        }
-    }
-}

src/net2/hosts.rs

@@ -1,42 +0,0 @@
-use async_std::sync::{Arc, Mutex};
-
-use fxhash::FxHashSet;
-use url::Url;
-
-/// Pointer to hosts class.
-pub type HostsPtr = Arc<Hosts>;
-
-/// Manages a store of network addresses.
-pub struct Hosts {
-    addrs: Mutex<Vec<Url>>,
-}
-
-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: &[Url]) -> bool {
-        let a_set: FxHashSet<_> = addrs.iter().cloned().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<Url>) {
-        if !self.contains(&addrs).await {
-            self.addrs.lock().await.extend(addrs)
-        }
-    }
-
-    /// Return the list of hosts.
-    pub async fn load_all(&self) -> Vec<Url> {
-        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/net2/message.rs

@@ -1,238 +0,0 @@
-use std::io;
-
-use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
-use log::debug;
-use url::Url;
-
-use crate::{
-    impl_vec,
-    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<Url>,
-}
-
-/// 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)? })
-    }
-}
-
-// CLEAN THIS
-impl Encodable for Url {
-    fn encode<S: io::Write>(&self, s: S) -> Result<usize> {
-        let mut len = 0;
-        len += self.to_string().encode(s)?;
-        Ok(len)
-    }
-}
-
-impl Decodable for Url {
-    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
-        let url: String = Decodable::decode(&mut d)?;
-        Ok(Self::parse(&url)?)
-    }
-}
-
-impl_vec!(Url);
-
-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>(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>(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/net2/message_subscriber.rs

@@ -1,308 +0,0 @@
-use async_std::sync::{Arc, Mutex};
-use std::{any::Any, io, io::Cursor};
-
-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/net2/mod.rs

@@ -1,101 +0,0 @@
-/// 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::{TcpTransport, TlsTransport, TorTransport, Transport};

src/net2/p2p.rs

@@ -1,242 +0,0 @@
-use async_std::sync::{Arc, Mutex};
-use std::fmt;
-
-use async_executor::Executor;
-use fxhash::{FxHashMap, FxHashSet};
-use log::debug;
-use serde_json::json;
-use url::Url;
-
-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, Transport,
-};
-
-/// List of channels that are awaiting connection.
-pub type PendingChannels = Mutex<FxHashSet<Url>>;
-/// List of connected channels.
-pub type ConnectedChannels<T> = Mutex<fxhash::FxHashMap<Url, Arc<Channel<T>>>>;
-/// Atomic pointer to p2p interface.
-pub type P2pPtr<T> = Arc<P2p<T>>;
-
-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<T: Transport> {
-    pending: PendingChannels,
-    channels: ConnectedChannels<T>,
-    channel_subscriber: SubscriberPtr<Result<ChannelPtr<T>>>,
-    // Used both internally and externally
-    stop_subscriber: SubscriberPtr<Error>,
-    hosts: HostsPtr,
-    protocol_registry: ProtocolRegistry<T>,
-
-    // We keep a reference to the sessions used for get info
-    session_manual: Mutex<Option<Arc<ManualSession<T>>>>,
-    session_inbound: Mutex<Option<Arc<InboundSession<T>>>>,
-    session_outbound: Mutex<Option<Arc<OutboundSession<T>>>>,
-
-    state: Mutex<P2pState>,
-
-    settings: SettingsPtr,
-}
-
-impl<T: Transport> P2p<T> {
-    /// 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
-            .as_ref()
-            .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<T>> {
-        self.session_manual.lock().await.as_ref().unwrap().clone()
-    }
-    pub async fn session_inbound(&self) -> Arc<InboundSession<T>> {
-        self.session_inbound.lock().await.as_ref().unwrap().clone()
-    }
-    pub async fn session_outbound(&self) -> Arc<OutboundSession<T>> {
-        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: Arc<Channel<T>>) {
-        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: Arc<Channel<T>>) {
-        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: &Url) -> bool {
-        self.channels.lock().await.contains_key(addr)
-    }
-
-    /// Add a channel to the list of pending channels.
-    pub async fn add_pending(&self, addr: Url) -> bool {
-        self.pending.lock().await.insert(addr)
-    }
-
-    /// Remove a channel from the list of pending channels.
-    pub async fn remove_pending(&self, addr: &Url) {
-        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<T> {
-        &self.protocol_registry
-    }
-
-    /// Subscribe to a channel.
-    pub async fn subscribe_channel(&self) -> Subscription<Result<ChannelPtr<T>>> {
-        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
-    }
-}

src/net2/protocol/mod.rs

@@ -1,71 +0,0 @@
-/// 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, Transport,
-};
-
-pub async fn register_default_protocols<T: Transport>(p2p: P2pPtr<T>) {
-    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/net2/protocol/protocol_address.rs

@@ -1,120 +0,0 @@
-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, Transport,
-    },
-    ProtocolBase, ProtocolBasePtr, ProtocolJobsManager, ProtocolJobsManagerPtr,
-};
-
-/// Defines address and get-address messages.
-pub struct ProtocolAddress<T: Transport> {
-    channel: ChannelPtr<T>,
-    addrs_sub: MessageSubscription<message::AddrsMessage>,
-    get_addrs_sub: MessageSubscription<message::GetAddrsMessage>,
-    hosts: HostsPtr,
-    jobsman: ProtocolJobsManagerPtr<T>,
-}
-
-impl<T: Transport> ProtocolAddress<T> {
-    /// 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<T>, p2p: P2pPtr<T>) -> 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<T: Transport> ProtocolBase for ProtocolAddress<T> {
-    /// 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/net2/protocol/protocol_base.rs

@@ -1,14 +0,0 @@
-use async_trait::async_trait;
-use smol::Executor;
-use std::sync::Arc;
-
-use crate::error::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/net2/protocol/protocol_jobs_manager.rs

@@ -1,70 +0,0 @@
-use async_std::sync::Mutex;
-use futures::Future;
-use log::*;
-use smol::Task;
-use std::sync::Arc;
-
-use crate::{error::Result, system::ExecutorPtr};
-
-use super::super::{ChannelPtr, Transport};
-
-/// Pointer to protocol jobs manager.
-pub type ProtocolJobsManagerPtr<T> = Arc<ProtocolJobsManager<T>>;
-
-/// 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<T: Transport> {
-    name: &'static str,
-    channel: ChannelPtr<T>,
-    tasks: Mutex<Vec<Task<Result<()>>>>,
-}
-
-impl<T: Transport> ProtocolJobsManager<T> {
-    /// Create a new protocol jobs manager.
-    pub fn new(name: &'static str, channel: ChannelPtr<T>) -> 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/net2/protocol/protocol_ping.rs

@@ -1,130 +0,0 @@
-use async_trait::async_trait;
-use log::{debug, error};
-use rand::Rng;
-use smol::Executor;
-use std::{sync::Arc, time::Instant};
-
-use crate::{
-    error::{Error, Result},
-    util::sleep,
-};
-
-use super::{
-    super::{
-        message, message_subscriber::MessageSubscription, ChannelPtr, P2pPtr, SettingsPtr,
-        Transport,
-    },
-    ProtocolBase, ProtocolBasePtr, ProtocolJobsManager, ProtocolJobsManagerPtr,
-};
-
-/// Defines ping and pong messages.
-pub struct ProtocolPing<T: Transport> {
-    channel: ChannelPtr<T>,
-    ping_sub: MessageSubscription<message::PingMessage>,
-    pong_sub: MessageSubscription<message::PongMessage>,
-    settings: SettingsPtr,
-    jobsman: ProtocolJobsManagerPtr<T>,
-}
-
-impl<T: Transport> ProtocolPing<T> {
-    /// Create a new ping-pong protocol.
-    pub async fn init(channel: ChannelPtr<T>, p2p: P2pPtr<T>) -> 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<T: Transport> ProtocolBase for ProtocolPing<T> {
-    /// 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/net2/protocol/protocol_registry.rs

@@ -1,60 +0,0 @@
-use async_std::sync::Mutex;
-use futures::future::BoxFuture;
-use log::debug;
-use std::future::Future;
-
-use super::{
-    super::{session::SessionBitflag, ChannelPtr, P2pPtr, Transport},
-    ProtocolBasePtr,
-};
-
-type Constructor<T> =
-    Box<dyn Fn(ChannelPtr<T>, P2pPtr<T>) -> BoxFuture<'static, ProtocolBasePtr> + Send + Sync>;
-
-pub struct ProtocolRegistry<T: Transport> {
-    protocol_constructors: Mutex<Vec<(SessionBitflag, Constructor<T>)>>,
-}
-
-impl<T: Transport> Default for ProtocolRegistry<T> {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl<T: Transport> ProtocolRegistry<T> {
-    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<T>, P2pPtr<T>) -> 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<T>,
-        p2p: P2pPtr<T>,
-    ) -> 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/net2/protocol/protocol_seed.rs

@@ -1,79 +0,0 @@
-use async_trait::async_trait;
-use log::debug;
-use smol::Executor;
-use std::sync::Arc;
-
-use crate::error::Result;
-
-use super::{
-    super::{message, ChannelPtr, HostsPtr, P2pPtr, SettingsPtr, Transport},
-    ProtocolBase, ProtocolBasePtr,
-};
-
-/// Implements the seed protocol.
-pub struct ProtocolSeed<T: Transport> {
-    channel: ChannelPtr<T>,
-    hosts: HostsPtr,
-    settings: SettingsPtr,
-}
-
-impl<T: Transport> ProtocolSeed<T> {
-    /// Create a new seed protocol.
-    pub async fn init(channel: ChannelPtr<T>, p2p: P2pPtr<T>) -> 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.clone() {
-            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<T: Transport> ProtocolBase for ProtocolSeed<T> {
-    /// 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/net2/protocol/protocol_version.rs

@@ -1,104 +0,0 @@
-use async_std::future::timeout;
-use log::*;
-use smol::Executor;
-use std::{sync::Arc, time::Duration};
-
-use crate::{Error, Result};
-
-use super::super::{
-    message, message_subscriber::MessageSubscription, ChannelPtr, SettingsPtr, Transport,
-};
-
-/// Implements the protocol version handshake sent out by nodes at the beginning
-/// of a connection.
-pub struct ProtocolVersion<T: Transport> {
-    channel: ChannelPtr<T>,
-    version_sub: MessageSubscription<message::VersionMessage>,
-    verack_sub: MessageSubscription<message::VerackMessage>,
-    settings: SettingsPtr,
-}
-
-impl<T: Transport> ProtocolVersion<T> {
-    /// 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<T>, 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/net2/session/inbound_session.rs

@@ -1,158 +0,0 @@
-use async_std::sync::Mutex;
-use async_trait::async_trait;
-use serde_json::json;
-use std::sync::{Arc, Weak};
-
-use async_executor::Executor;
-use fxhash::FxHashMap;
-use log::{error, info};
-use url::Url;
-
-use crate::{
-    error::{Error, Result},
-    system::{StoppableTask, StoppableTaskPtr},
-};
-
-use super::{
-    super::{Acceptor, AcceptorPtr, ChannelPtr, P2p, Transport},
-    Session, SessionBitflag, SESSION_INBOUND,
-};
-
-struct InboundInfo<T: Transport> {
-    channel: ChannelPtr<T>,
-}
-
-impl<T: Transport> InboundInfo<T> {
-    async fn get_info(&self) -> serde_json::Value {
-        self.channel.get_info().await
-    }
-}
-
-/// Defines inbound connections session.
-pub struct InboundSession<T: Transport> {
-    p2p: Weak<P2p<T>>,
-    acceptor: AcceptorPtr<T>,
-    accept_task: StoppableTaskPtr,
-    connect_infos: Mutex<FxHashMap<Url, InboundInfo<T>>>,
-}
-
-impl<T: Transport> InboundSession<T> {
-    /// Create a new inbound session.
-    pub fn new(p2p: Weak<P2p<T>>) -> 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.clone() {
-            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: Url,
-        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<T>,
-        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<T>) {
-        let key = channel.address();
-        self.connect_infos
-            .lock()
-            .await
-            .insert(key.clone(), 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<T: Transport> Session<T> for InboundSession<T> {
-    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<T>> {
-        self.p2p.upgrade().unwrap()
-    }
-
-    fn selector_id(&self) -> SessionBitflag {
-        SESSION_INBOUND
-    }
-}

src/net2/session/manual_session.rs

@@ -1,148 +0,0 @@
-use async_std::sync::Mutex;
-use async_trait::async_trait;
-use std::sync::{Arc, Weak};
-
-use async_executor::Executor;
-use log::*;
-use serde_json::json;
-use url::Url;
-
-use crate::{
-    error::{Error, Result},
-    system::{StoppableTask, StoppableTaskPtr},
-    util::sleep,
-};
-
-use super::{
-    super::{Connector, P2p, Transport},
-    Session, SessionBitflag, SESSION_MANUAL,
-};
-
-pub struct ManualSession<T: Transport> {
-    p2p: Weak<P2p<T>>,
-    connect_slots: Mutex<Vec<StoppableTaskPtr>>,
-}
-
-impl<T: Transport> ManualSession<T> {
-    /// Create a new inbound session.
-    pub fn new(p2p: Weak<P2p<T>>) -> 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: &Url, executor: Arc<Executor<'_>>) {
-        let task = StoppableTask::new();
-
-        task.clone().start(
-            self.clone().channel_connect_loop(addr.clone(), 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: Url,
-        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.clone()).await;
-
-            info!(target: "net", "Connecting to manual outbound [{}]", &addr);
-
-            match connector.connect(addr.clone()).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<T: Transport> Session<T> for ManualSession<T> {
-    async fn get_info(&self) -> serde_json::Value {
-        json!({
-            "key": 110
-        })
-    }
-
-    fn p2p(&self) -> Arc<P2p<T>> {
-        self.p2p.upgrade().unwrap()
-    }
-
-    fn selector_id(&self) -> SessionBitflag {
-        SESSION_MANUAL
-    }
-}

src/net2/session/mod.rs

@@ -1,143 +0,0 @@
-use async_trait::async_trait;
-use log::debug;
-use smol::Executor;
-use std::sync::Arc;
-
-use crate::error::Result;
-
-use super::{p2p::P2pPtr, protocol::ProtocolVersion, transport::Transport, 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<T: Transport>(p2p: P2pPtr<T>, channel: ChannelPtr<T>) {
-    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<T: Transport>: 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<T>,
-        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<T>>,
-        channel: ChannelPtr<T>,
-        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<T>;
-
-    fn selector_id(&self) -> u32;
-}

src/net2/session/outbound_session.rs

@@ -1,257 +0,0 @@
-use async_std::sync::Mutex;
-use std::{
-    fmt,
-    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 url::Url;
-
-use crate::{
-    error::{Error, Result},
-    system::{StoppableTask, StoppableTaskPtr},
-};
-
-use super::{
-    super::{ChannelPtr, Connector, P2p, Transport},
-    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<T: Transport> {
-    addr: Option<Url>,
-    channel: Option<ChannelPtr<T>>,
-    state: OutboundState,
-}
-
-impl<T: Transport> OutboundInfo<T> {
-    async fn get_info(&self) -> serde_json::Value {
-        let addr = match self.addr.clone() {
-            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<T: Transport> Default for OutboundInfo<T> {
-    fn default() -> Self {
-        Self { addr: None, channel: None, state: OutboundState::Open }
-    }
-}
-
-/// Defines outbound connections session.
-pub struct OutboundSession<T: Transport> {
-    p2p: Weak<P2p<T>>,
-    connect_slots: Mutex<Vec<StoppableTaskPtr>>,
-    slot_info: Mutex<Vec<OutboundInfo<T>>>,
-}
-
-impl<T: Transport> OutboundSession<T> {
-    /// Create a new outbound session.
-    pub fn new(p2p: Weak<P2p<T>>) -> 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.clone());
-                info.state = OutboundState::Pending;
-            }
-
-            match connector.connect(addr.clone()).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<Url> {
-        let p2p = self.p2p();
-        let self_inbound_addr = p2p.settings().external_addr.clone();
-
-        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.clone()).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: &Url, inbound_addr: &Option<Url>) -> bool {
-        match inbound_addr {
-            Some(inbound_addr) => inbound_addr == addr,
-            // No inbound listening address configured
-            None => false,
-        }
-    }
-}
-
-#[async_trait]
-impl<T: Transport> Session<T> for OutboundSession<T> {
-    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<T>> {
-        self.p2p.upgrade().unwrap()
-    }
-
-    fn selector_id(&self) -> SessionBitflag {
-        SESSION_OUTBOUND
-    }
-}

src/net2/session/seed_session.rs

@@ -1,152 +0,0 @@
-use async_std::future::timeout;
-use async_trait::async_trait;
-use std::{
-    sync::{Arc, Weak},
-    time::Duration,
-};
-
-use async_executor::Executor;
-use log::*;
-use serde_json::json;
-use url::Url;
-
-use crate::error::{Error, Result};
-
-use super::super::{
-    session::{Session, SessionBitflag, SESSION_SEED},
-    Connector, P2p, Transport,
-};
-
-/// Defines seed connections session.
-pub struct SeedSession<T: Transport> {
-    p2p: Weak<P2p<T>>,
-}
-
-impl<T: Transport> SeedSession<T> {
-    /// Create a new seed session instance.
-    pub fn new(p2p: Weak<P2p<T>>) -> 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.clone(), 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: Url,
-        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.clone()).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<T: Transport> Session<T> for SeedSession<T> {
-    async fn get_info(&self) -> serde_json::Value {
-        json!({
-            "key": 110
-        })
-    }
-
-    fn p2p(&self) -> Arc<P2p<T>> {
-        self.p2p.upgrade().unwrap()
-    }
-
-    fn selector_id(&self) -> SessionBitflag {
-        SESSION_SEED
-    }
-}

src/net2/settings.rs

@@ -1,40 +0,0 @@
-use std::sync::Arc;
-
-use url::Url;
-
-/// Atomic pointer to network settings.
-pub type SettingsPtr = Arc<Settings>;
-
-/// Defines the network settings.
-#[derive(Clone)]
-pub struct Settings {
-    pub inbound: Option<Url>,
-    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<Url>,
-    pub peers: Vec<Url>,
-    pub seeds: Vec<Url>,
-}
-
-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/net2/transport.rs

@@ -1,55 +0,0 @@
-use async_std::sync::Arc;
-use std::error::Error;
-
-use async_trait::async_trait;
-use futures::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: Sync + Send + 'static + Clone {
-    type Acceptor: Sync + Send;
-    type Connector: Sync + Send + AsyncRead + AsyncWrite;
-
-    type Error: Error;
-
-    type Listener: Future<Output = Result<Self::Acceptor, TransportError<Self::Error>>>
-        + Sync
-        + Send;
-    type Dial: Future<Output = Result<Self::Connector, TransportError<Self::Error>>> + Sync + Send;
-
-    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;
-
-    fn new(ttl: Option<u32>, backlog: i32) -> Self;
-
-    async fn accept(
-        listener: Arc<Self::Acceptor>,
-    ) -> Result<Self::Connector, TransportError<Self::Error>>;
-}
-
-#[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),
-}
-unsafe impl<TErr> Sync for TransportError<TErr> {}
-unsafe impl<TErr> Send for TransportError<TErr> {}

src/net2/transport/tcp.rs

@@ -1,114 +0,0 @@
-use async_std::{
-    net::{TcpListener, TcpStream},
-    sync::Arc,
-};
-use std::{io, net::SocketAddr, pin::Pin};
-
-use async_trait::async_trait;
-use futures::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,
-}
-
-#[async_trait]
-impl Transport for TcpTransport {
-    type Acceptor = TcpListener;
-    type Connector = TcpStream;
-
-    type Error = io::Error;
-
-    type Listener = Pin<
-        Box<dyn Future<Output = Result<Self::Acceptor, TransportError<Self::Error>>> + Send + Sync>,
-    >;
-    type Dial = Pin<
-        Box<
-            dyn Future<Output = Result<Self::Connector, TransportError<Self::Error>>> + Send + Sync,
-        >,
-    >;
-
-    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)))
-    }
-
-    fn new(ttl: Option<u32>, backlog: i32) -> Self {
-        Self { ttl, backlog }
-    }
-
-    async fn accept(
-        listener: Arc<Self::Acceptor>,
-    ) -> Result<Self::Connector, TransportError<Self::Error>> {
-        Ok(listener.accept().await?.0)
-    }
-}
-
-impl TcpTransport {
-    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, TransportError<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, TransportError<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(TransportError::Other(err)),
-        };
-
-        let stream = TcpStream::from(std::net::TcpStream::from(socket));
-        Ok(stream)
-    }
-}

src/net2/transport/tls.rs

@@ -1,217 +0,0 @@
-use async_std::net::{TcpListener, TcpStream};
-use std::{io, net::SocketAddr, pin::Pin, sync::Arc, time::SystemTime};
-
-use async_trait::async_trait;
-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>,
-}
-
-#[async_trait]
-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, TransportError<Self::Error>>> + Send + Sync>,
-    >;
-    type Dial = Pin<
-        Box<
-            dyn Future<Output = Result<Self::Connector, TransportError<Self::Error>>> + Send + Sync,
-        >,
-    >;
-
-    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)))
-    }
-
-    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 }
-    }
-
-    async fn accept(
-        listener: Arc<Self::Acceptor>,
-    ) -> Result<Self::Connector, TransportError<Self::Error>> {
-        let stream = listener.1.accept().await?.0;
-        Ok(listener.0.accept(stream).await?.into())
-    }
-}
-
-impl TlsTransport {
-    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), TransportError<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>, TransportError<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(TransportError::Other(err)),
-        };
-
-        let stream = TcpStream::from(std::net::TcpStream::from(socket));
-        let stream = connector.connect(server_name, stream).await?;
-        Ok(TlsStream::Client(stream))
-    }
-}

src/net2/transport/tor.rs

@@ -1,278 +0,0 @@
-use async_std::{
-    net::{TcpListener, TcpStream},
-    sync::Arc,
-};
-use std::{
-    io,
-    io::{BufRead, BufReader, Write},
-    net::SocketAddr,
-    pin::Pin,
-    time::Duration,
-};
-
-use async_trait::async_trait;
-use fast_socks5::{
-    client::{Config, Socks5Stream},
-    Result, SocksError,
-};
-use futures::prelude::*;
-
-use regex::Regex;
-use socket2::{Domain, Socket, Type};
-
-use url::Url;
-
-use super::{Transport, TransportError};
-
-/// 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_t(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_t(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_t(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>, TransportError<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
-        };
-        // FIXME
-        Ok(stream.unwrap())
-    }
-
-    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, TransportError<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)))
-    }
-}
-
-#[async_trait]
-impl Transport for TorTransport {
-    type Acceptor = TcpListener;
-    type Connector = Socks5Stream<TcpStream>;
-
-    type Error = TorError;
-
-    type Listener = Pin<
-        Box<dyn Future<Output = Result<Self::Acceptor, TransportError<Self::Error>>> + Send + Sync>,
-    >;
-    type Dial = Pin<
-        Box<
-            dyn Future<Output = Result<Self::Connector, TransportError<Self::Error>>> + Send + Sync,
-        >,
-    >;
-
-    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)))
-    }
-    fn new(_ttl: Option<u32>, _backlog: i32) -> Self {
-        unimplemented!()
-    }
-
-    async fn accept(
-        _listener: Arc<Self::Acceptor>,
-    ) -> Result<Self::Connector, TransportError<Self::Error>> {
-        unimplemented!()
-    }
-}