Simplify connector type, drop .boxed_connector() method

Author: fafhrd91

CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [0.3.1] - 2022-01-10
+
+* Simplify connector type, drop .boxed_connector() method
+
+* Remove openssl and rustls features
+
 ## [0.3.0] - 2021-12-30
 
 * Upgrade to ntex 0.5.0

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-redis"
-version = "0.3.0"
+version = "0.3.1"
 authors = ["ntex contributors <[email protected]>"]
 description = "Redis client"
 documentation = "https://docs.rs/ntex-redis"
@@ -11,17 +11,8 @@ license = "MIT"
 exclude = [".gitignore", ".travis.yml", ".cargo/config"]
 edition = "2018"
 
-[features]
-default = []
-
-# openssl
-openssl = ["ntex/openssl"]
-
-# rustls support
-rustls = ["ntex/rustls"]
-
 [dependencies]
-ntex = "0.5.0"
+ntex = "0.5.8"
 itoa = "0.4.5"
 btoi = "0.4.2"
 log = "0.4"
@@ -30,3 +21,4 @@ derive_more = "0.99"
 [dev-dependencies]
 rand = "0.8"
 env_logger = "0.9"
+ntex = { version = "0.5", features = ["tokio"] }

src/client.rs

@@ -1,7 +1,7 @@
 use std::collections::VecDeque;
 use std::{cell::RefCell, fmt, future::Future, pin::Pin, rc::Rc, task::Context, task::Poll};
 
-use ntex::io::{utils::OnDisconnect, IoBoxed, IoRef, RecvError};
+use ntex::io::{IoBoxed, IoRef, OnDisconnect, RecvError};
 use ntex::util::{poll_fn, ready, Either, Ready};
 use ntex::{channel::pool, service::Service};
 
@@ -146,11 +146,10 @@ pub struct CommandResult {
 impl Future for CommandResult {
     type Output = Result<Response, Error>;
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match Pin::new(&mut self.rx).poll(cx) {
-            Poll::Ready(Ok(res)) => Poll::Ready(res),
-            Poll::Ready(Err(_)) => Poll::Ready(Err(Error::PeerGone(None))),
-            Poll::Pending => Poll::Pending,
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match ready!(self.rx.poll_recv(cx)) {
+            Ok(res) => Poll::Ready(res),
+            Err(_) => Poll::Ready(Err(Error::PeerGone(None))),
         }
     }
 }

src/connector.rs

@@ -1,15 +1,9 @@
 use std::future::Future;
 
 use ntex::connect::{self, Address, Connect, Connector};
-use ntex::io::{utils::Boxed, IoBoxed};
+use ntex::io::IoBoxed;
 use ntex::{service::Service, time::Seconds, util::ByteString, util::PoolId, util::PoolRef};
 
-#[cfg(feature = "openssl")]
-use ntex::connect::openssl::{self, SslConnector};
-
-#[cfg(feature = "rustls")]
-use ntex::connect::rustls::{self, ClientConfig};
-
 use super::errors::ConnectError;
 use super::{cmd, Client, SimpleClient};
 
@@ -27,17 +21,20 @@ where
 {
     #[allow(clippy::new_ret_no_self)]
     /// Create new redis connector
-    pub fn new(address: A) -> RedisConnector<A, Boxed<Connector<A>, Connect<A>>> {
+    pub fn new(address: A) -> RedisConnector<A, Connector<A>> {
         RedisConnector {
             address,
             passwords: Vec::new(),
-            connector: Connector::default().seal(),
+            connector: Connector::default(),
             pool: PoolId::P7.pool_ref(),
         }
     }
 }
 
-impl<A, T> RedisConnector<A, T> {
+impl<A, T> RedisConnector<A, T>
+where
+    A: Address + Clone,
+{
     /// Add redis auth password
     pub fn password<U>(mut self, password: U) -> Self
     where
@@ -58,23 +55,10 @@ impl<A, T> RedisConnector<A, T> {
     }
 
     /// Use custom connector
-    pub fn connector<Io, U>(self, connector: U) -> RedisConnector<A, Boxed<U, Connect<A>>>
-    where
-        U: Service<Connect<A>, Response = Io, Error = connect::ConnectError>,
-        IoBoxed: From<Io>,
-    {
-        RedisConnector {
-            connector: Boxed::new(connector),
-            address: self.address,
-            passwords: self.passwords,
-            pool: self.pool,
-        }
-    }
-
-    /// Use custom boxed connector
-    pub fn boxed_connector<U>(self, connector: U) -> RedisConnector<A, U>
+    pub fn connector<U>(self, connector: U) -> RedisConnector<A, U>
     where
-        U: Service<Connect<A>, Response = IoBoxed, Error = connect::ConnectError>,
+        U: Service<Connect<A>, Error = connect::ConnectError>,
+        IoBoxed: From<U::Response>,
     {
         RedisConnector {
             connector,
@@ -83,90 +67,48 @@ impl<A, T> RedisConnector<A, T> {
             pool: self.pool,
         }
     }
-
-    #[cfg(feature = "openssl")]
-    /// Use openssl connector.
-    pub fn openssl(
-        self,
-        connector: SslConnector,
-    ) -> RedisConnector<A, Boxed<openssl::Connector<A>, Connect<A>>> {
-        RedisConnector {
-            address: self.address,
-            passwords: self.passwords,
-            connector: Boxed::new(openssl::Connector::new(connector)),
-            pool: self.pool,
-        }
-    }
-
-    #[cfg(feature = "rustls")]
-    /// Use rustls connector.
-    pub fn rustls(
-        self,
-        config: ClientConfig,
-    ) -> RedisConnector<A, Boxed<rustls::Connector<A>, Connect<A>>> {
-        RedisConnector {
-            address: self.address,
-            passwords: self.passwords,
-            connector: Boxed::new(rustls::Connector::new(config)),
-            pool: self.pool,
-        }
-    }
 }
 
 impl<A, T> RedisConnector<A, T>
 where
     A: Address + Clone,
-    T: Service<Connect<A>, Response = IoBoxed, Error = connect::ConnectError>,
+    T: Service<Connect<A>, Error = connect::ConnectError>,
+    IoBoxed: From<T::Response>,
 {
-    /// Connect to redis server and create shared client
-    pub fn connect(&self) -> impl Future<Output = Result<Client, ConnectError>> {
+    fn _connect(&self) -> impl Future<Output = Result<IoBoxed, ConnectError>> {
         let pool = self.pool;
         let passwords = self.passwords.clone();
         let fut = self.connector.call(Connect::new(self.address.clone()));
 
         async move {
-            let io = fut.await?;
+            let io = IoBoxed::from(fut.await?);
             io.set_memory_pool(pool);
             io.set_disconnect_timeout(Seconds::ZERO.into());
 
-            let client = Client::new(io);
-
             if passwords.is_empty() {
-                Ok(client)
+                Ok(io)
             } else {
+                let client = SimpleClient::new(io);
+
                 for password in passwords {
                     if client.exec(cmd::Auth(password)).await? {
-                        return Ok(client);
+                        return Ok(client.into_inner());
                     }
                 }
                 Err(ConnectError::Unauthorized)
             }
         }
     }
 
+    /// Connect to redis server and create shared client
+    pub fn connect(&self) -> impl Future<Output = Result<Client, ConnectError>> {
+        let fut = self._connect();
+        async move { fut.await.map(|io| Client::new(io)) }
+    }
+
     /// Connect to redis server and create simple client
     pub fn connect_simple(&self) -> impl Future<Output = Result<SimpleClient, ConnectError>> {
-        let pool = self.pool;
-        let passwords = self.passwords.clone();
-        let fut = self.connector.call(Connect::new(self.address.clone()));
-
-        async move {
-            let io = fut.await?;
-            io.set_memory_pool(pool);
-            io.set_disconnect_timeout(Seconds::ZERO.into());
-
-            let client = SimpleClient::new(io);
-
-            if passwords.is_empty() {
-                Ok(client)
-            } else {
-                for password in passwords {
-                    if client.exec(cmd::Auth(password)).await? {
-                        return Ok(client);
-                    }
-                }
-                Err(ConnectError::Unauthorized)
-            }
-        }
+        let fut = self._connect();
+        async move { fut.await.map(|io| SimpleClient::new(io)) }
     }
 }

src/errors.rs

@@ -64,7 +64,7 @@ pub enum CommandError {
     Error(ByteString),
 
     /// A command response parse error
-    #[display(fmt = "Command output error: {}", _0)]
+    #[display(fmt = "Command output parse error: {}", _0)]
     Output(&'static str, Response),
 
     /// Redis protocol level errors

src/lib.rs

@@ -29,6 +29,8 @@
 //!     Ok(())
 //! }
 //! ```
+#![allow(clippy::return_self_not_must_use)]
+
 mod client;
 pub mod cmd;
 pub mod codec;

src/simple.rs

@@ -45,4 +45,8 @@ impl SimpleClient {
         })
         .await
     }
+
+    pub(crate) fn into_inner(self) -> IoBoxed {
+        self.io
+    }
 }