net3: create wrapper for stream returned by Transport connections

src/net3/channel.rs

@@ -1,14 +1,10 @@
-use async_std::{
-    net::TcpStream,
-    sync::{Arc, Mutex},
-};
+use async_std::sync::{Arc, Mutex};
 use std::sync::atomic::{AtomicBool, Ordering};
 
 use futures::{
     io::{ReadHalf, WriteHalf},
-    AsyncRead, AsyncReadExt, AsyncWrite,
+    AsyncReadExt,
 };
-use futures_rustls::TlsStream;
 use log::{debug, error, info};
 use rand::Rng;
 use serde_json::json;
@@ -23,6 +19,7 @@ use crate::{
 use super::{
     message,
     message_subscriber::{MessageSubscription, MessageSubsystem},
+    TransportStream,
 };
 
 /// Atomic pointer to async channel.
@@ -58,15 +55,10 @@ impl ChannelInfo {
     }
 }
 
-pub trait Stream: AsyncWrite + AsyncRead + Unpin + Send + Sync {}
-
-impl Stream for TcpStream {}
-impl<T: Stream> Stream for TlsStream<T> {}
-
 /// Async channel for communication between nodes.
 pub struct Channel {
-    reader: Mutex<ReadHalf<Box<dyn Stream>>>,
-    writer: Mutex<WriteHalf<Box<dyn Stream>>>,
+    reader: Mutex<ReadHalf<Box<dyn TransportStream>>>,
+    writer: Mutex<WriteHalf<Box<dyn TransportStream>>>,
     address: Url,
     message_subsystem: MessageSubsystem,
     stop_subscriber: SubscriberPtr<Error>,
@@ -79,7 +71,7 @@ impl Channel {
     /// Sets up a new channel. Creates a reader and writer TCP stream and
     /// summons the message subscriber subsystem. Performs a network
     /// handshake on the subsystem dispatchers.
-    pub async fn new(stream: Box<dyn Stream>, address: Url) -> Arc<Self> {
+    pub async fn new(stream: Box<dyn TransportStream>, address: Url) -> Arc<Self> {
         let (reader, writer) = stream.split();
         let reader = Mutex::new(reader);
         let writer = Mutex::new(writer);

src/net3/mod.rs

@@ -89,7 +89,7 @@ pub mod settings;
 pub mod transport;
 
 pub use acceptor::{Acceptor, AcceptorPtr};
-pub use channel::{Channel, ChannelPtr, Stream};
+pub use channel::{Channel, ChannelPtr};
 pub use connector::Connector;
 pub use hosts::{Hosts, HostsPtr};
 pub use message::Message;
@@ -98,4 +98,4 @@ 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, TorTransport, Transport};
+pub use transport::{TcpTransport, TorTransport, Transport, TransportStream};

src/net3/transport.rs

@@ -1,4 +1,5 @@
 use async_trait::async_trait;
+
 use futures::prelude::*;
 use futures_rustls::{TlsAcceptor, TlsStream};
 use url::Url;
@@ -14,6 +15,9 @@ pub use tcp::TcpTransport;
 mod tor;
 pub use tor::TorTransport;
 
+/// This used as wrapper for stream return by dial function inside Transport trait
+pub trait TransportStream: AsyncWrite + AsyncRead + Unpin + Send + Sync {}
+
 /// The `Transport` trait serves as a base for implementing transport protocols.
 /// Base transports can optionally be upgraded with TLS in order to support encryption.
 /// The implementation of our TLS authentication can be found in the [`upgrade_tls`] module.

src/net3/transport/tcp.rs

@@ -7,9 +7,12 @@ use log::debug;
 use socket2::{Domain, Socket, Type};
 use url::Url;
 
-use super::{TlsUpgrade, Transport};
+use super::{TlsUpgrade, Transport, TransportStream};
 use crate::{Error, Result};
 
+impl TransportStream for TcpStream {}
+impl<T: TransportStream> TransportStream for TlsStream<T> {}
+
 #[derive(Copy, Clone)]
 pub struct TcpTransport {
     /// TTL to set for opened sockets, or `None` for default