refactor: AsyncReadWrite (#822)

Author: Devdutt Shenoi

rumqttc/CHANGELOG.md

@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+* rename `N` as `AsyncReadWrite` to describe usage.
+
 ### Deprecated
 
 ### Removed

rumqttc/src/eventloop.rs

@@ -2,7 +2,7 @@ use crate::{framed::Network, Transport};
 use crate::{Incoming, MqttState, NetworkOptions, Packet, Request, StateError};
 use crate::{MqttOptions, Outgoing};
 
-use crate::framed::N;
+use crate::framed::AsyncReadWrite;
 use crate::mqttbytes::v4::*;
 use flume::{bounded, Receiver, Sender};
 use tokio::net::{lookup_host, TcpSocket, TcpStream};
@@ -369,7 +369,7 @@ async fn network_connect(
         _ => options.broker_address(),
     };
 
-    let tcp_stream: Box<dyn N> = {
+    let tcp_stream: Box<dyn AsyncReadWrite> = {
         #[cfg(feature = "proxy")]
         match options.proxy() {
             Some(proxy) => proxy.connect(&domain, port, network_options).await?,

rumqttc/src/framed.rs

@@ -10,7 +10,7 @@ use std::io;
 /// appropriate to achieve performance
 pub struct Network {
     /// Socket for IO
-    socket: Box<dyn N>,
+    socket: Box<dyn AsyncReadWrite>,
     /// Buffered reads
     read: BytesMut,
     /// Maximum packet size
@@ -20,8 +20,8 @@ pub struct Network {
 }
 
 impl Network {
-    pub fn new(socket: impl N + 'static, max_incoming_size: usize) -> Network {
-        let socket = Box::new(socket) as Box<dyn N>;
+    pub fn new(socket: impl AsyncReadWrite + 'static, max_incoming_size: usize) -> Network {
+        let socket = Box::new(socket) as Box<dyn AsyncReadWrite>;
         Network {
             socket,
             read: BytesMut::with_capacity(10 * 1024),
@@ -117,5 +117,5 @@ impl Network {
     }
 }
 
-pub trait N: AsyncRead + AsyncWrite + Send + Unpin {}
-impl<T> N for T where T: AsyncRead + AsyncWrite + Send + Unpin {}
+pub trait AsyncReadWrite: AsyncRead + AsyncWrite + Send + Unpin {}
+impl<T> AsyncReadWrite for T where T: AsyncRead + AsyncWrite + Send + Unpin {}

rumqttc/src/proxy.rs

@@ -1,5 +1,5 @@
 use crate::eventloop::socket_connect;
-use crate::framed::N;
+use crate::framed::AsyncReadWrite;
 use crate::NetworkOptions;
 
 use std::io;
@@ -46,10 +46,10 @@ impl Proxy {
         broker_addr: &str,
         broker_port: u16,
         network_options: NetworkOptions,
-    ) -> Result<Box<dyn N>, ProxyError> {
+    ) -> Result<Box<dyn AsyncReadWrite>, ProxyError> {
         let proxy_addr = format!("{}:{}", self.addr, self.port);
 
-        let tcp: Box<dyn N> = Box::new(socket_connect(proxy_addr, network_options).await?);
+        let tcp: Box<dyn AsyncReadWrite> = Box::new(socket_connect(proxy_addr, network_options).await?);
         let mut tcp = match self.ty {
             ProxyType::Http => tcp,
             #[cfg(any(feature = "use-rustls", feature = "use-native-tls"))]
@@ -67,7 +67,7 @@ impl ProxyAuth {
         self,
         host: &str,
         port: u16,
-        tcp_stream: &mut Box<dyn N>,
+        tcp_stream: &mut Box<dyn AsyncReadWrite>,
     ) -> Result<(), ProxyError> {
         match self {
             Self::None => async_http_proxy::http_connect_tokio(tcp_stream, host, port).await?,

rumqttc/src/tls.rs

@@ -16,7 +16,7 @@ use std::io::{BufReader, Cursor};
 #[cfg(feature = "use-rustls")]
 use std::sync::Arc;
 
-use crate::framed::N;
+use crate::framed::AsyncReadWrite;
 use crate::TlsConfiguration;
 
 #[cfg(feature = "use-native-tls")]
@@ -166,9 +166,9 @@ pub async fn tls_connect(
     addr: &str,
     _port: u16,
     tls_config: &TlsConfiguration,
-    tcp: Box<dyn N>,
-) -> Result<Box<dyn N>, Error> {
-    let tls: Box<dyn N> = match tls_config {
+    tcp: Box<dyn AsyncReadWrite>,
+) -> Result<Box<dyn AsyncReadWrite>, Error> {
+    let tls: Box<dyn AsyncReadWrite> = match tls_config {
         #[cfg(feature = "use-rustls")]
         TlsConfiguration::Simple { .. } | TlsConfiguration::Rustls(_) => {
             let connector = rustls_connector(tls_config).await?;

rumqttc/src/v5/eventloop.rs

@@ -2,7 +2,7 @@ use super::framed::Network;
 use super::mqttbytes::v5::*;
 use super::{Incoming, MqttOptions, MqttState, Outgoing, Request, StateError, Transport};
 use crate::eventloop::socket_connect;
-use crate::framed::N;
+use crate::framed::AsyncReadWrite;
 
 use flume::{bounded, Receiver, Sender};
 use tokio::select;
@@ -304,7 +304,7 @@ async fn network_connect(options: &MqttOptions) -> Result<Network, ConnectionErr
         _ => options.broker_address(),
     };
 
-    let tcp_stream: Box<dyn N> = {
+    let tcp_stream: Box<dyn AsyncReadWrite> = {
         #[cfg(feature = "proxy")]
         match options.proxy() {
             Some(proxy) => {

rumqttc/src/v5/framed.rs

@@ -1,5 +1,7 @@
 use bytes::BytesMut;
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+use crate::framed::AsyncReadWrite;
 
 use super::mqttbytes;
 use super::mqttbytes::v5::{Connect, Login, Packet};
@@ -11,7 +13,7 @@ use std::io;
 /// appropriate to achieve performance
 pub struct Network {
     /// Socket for IO
-    socket: Box<dyn N>,
+    socket: Box<dyn AsyncReadWrite>,
     /// Buffered reads
     read: BytesMut,
     /// Maximum packet size
@@ -21,8 +23,8 @@ pub struct Network {
 }
 
 impl Network {
-    pub fn new(socket: impl N + 'static, max_incoming_size: Option<usize>) -> Network {
-        let socket = Box::new(socket) as Box<dyn N>;
+    pub fn new(socket: impl AsyncReadWrite + 'static, max_incoming_size: Option<usize>) -> Network {
+        let socket = Box::new(socket) as Box<dyn AsyncReadWrite>;
         Network {
             socket,
             read: BytesMut::with_capacity(10 * 1024),
@@ -127,6 +129,3 @@ impl Network {
         Ok(())
     }
 }
-
-pub trait N: AsyncRead + AsyncWrite + Send + Unpin {}
-impl<T> N for T where T: AsyncRead + AsyncWrite + Send + Unpin {}