upgrade to ntex 0.5

Author: fafhrd91

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.3.0-b.0] - 2021-12-19
+
+* upgrade to ntex 0.5
+
 ## [0.2.4] - 2021-12-02
 
 * Add memory pools support

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-redis"
-version = "0.2.4"
+version = "0.3.0-b.0"
 authors = ["ntex contributors <[email protected]>"]
 description = "Redis client"
 documentation = "https://docs.rs/ntex-redis"
@@ -21,7 +21,7 @@ openssl = ["ntex/openssl"]
 rustls = ["ntex/rustls"]
 
 [dependencies]
-ntex = "0.4.11"
+ntex = "0.5.0-b.0"
 itoa = "0.4.5"
 btoi = "0.4.2"
 log = "0.4"

src/client.rs

@@ -1,8 +1,9 @@
 use std::collections::VecDeque;
 use std::{cell::RefCell, fmt, future::Future, pin::Pin, rc::Rc, task::Context, task::Poll};
 
+use ntex::io::{IoBoxed, IoRef, OnDisconnect};
 use ntex::util::{poll_fn, Either, Ready};
-use ntex::{channel::pool, framed::State, service::Service};
+use ntex::{channel::pool, service::Service};
 
 use super::cmd::Command;
 use super::codec::{Codec, Request, Response};
@@ -13,20 +14,21 @@ type Queue = Rc<RefCell<VecDeque<pool::Sender<Result<Response, Error>>>>>;
 #[derive(Clone)]
 /// Shared redis client
 pub struct Client {
-    state: State,
+    io: IoRef,
     queue: Queue,
+    disconnect: OnDisconnect,
     pool: pool::Pool<Result<Response, Error>>,
 }
 
 impl Client {
-    pub(crate) fn new(state: State) -> Self {
+    pub(crate) fn new(io: IoBoxed) -> Self {
         let queue: Queue = Rc::new(RefCell::new(VecDeque::new()));
 
         // read redis response task
-        let state2 = state.clone();
+        let io_ref = io.get_ref();
         let queue2 = queue.clone();
         ntex::rt::spawn(async move {
-            let read = state2.read();
+            let read = io.read();
 
             poll_fn(|cx| {
                 loop {
@@ -36,7 +38,7 @@ impl Client {
                                 let _ = tx.send(Err(e));
                             }
                             queue2.borrow_mut().clear();
-                            state2.shutdown_io();
+                            let _ = io.poll_shutdown(cx);
                             return Poll::Ready(());
                         }
                         Ok(Some(item)) => {
@@ -50,18 +52,24 @@ impl Client {
                     }
                 }
 
-                if !state2.is_open() {
+                if io.is_closed() {
                     return Poll::Ready(());
                 }
-                state2.register_dispatcher(cx.waker());
-                Poll::Pending
+                if let Err(_) = read.poll_read_ready(cx) {
+                    Poll::Ready(())
+                } else {
+                    Poll::Pending
+                }
             })
             .await
         });
 
+        let disconnect = io_ref.on_disconnect();
+
         Client {
-            state,
             queue,
+            disconnect,
+            io: io_ref,
             pool: pool::new(),
         }
     }
@@ -71,7 +79,7 @@ impl Client {
     where
         T: Command,
     {
-        let is_open = self.state.is_open();
+        let is_open = !self.io.is_closed();
         let fut = self.call(cmd.to_request());
 
         async move {
@@ -93,7 +101,7 @@ impl Client {
 
     /// Returns true if underlying transport is connected to redis
     pub fn is_connected(&self) -> bool {
-        self.state.is_open()
+        !self.io.is_closed()
     }
 }
 
@@ -103,16 +111,16 @@ impl Service for Client {
     type Error = Error;
     type Future = Either<CommandResult, Ready<Response, Error>>;
 
-    fn poll_ready(&self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        if !self.state.is_open() {
+    fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        if self.disconnect.poll_ready(cx).is_ready() {
             Poll::Ready(Err(Error::Disconnected))
         } else {
             Poll::Ready(Ok(()))
         }
     }
 
     fn call(&self, req: Request) -> Self::Future {
-        if let Err(e) = self.state.write().encode(req, &Codec) {
+        if let Err(e) = self.io.write().encode(req, &Codec) {
             Either::Right(Ready::Err(e))
         } else {
             let (tx, rx) = self.pool.channel();
@@ -125,7 +133,7 @@ impl Service for Client {
 impl fmt::Debug for Client {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Client")
-            .field("connected", &self.state.is_open())
+            .field("connected", &!self.io.is_closed())
             .finish()
     }
 }

src/connector.rs

@@ -1,8 +1,7 @@
-use std::{cell::RefCell, future::Future, rc::Rc};
+use std::future::Future;
 
-use ntex::codec::{AsyncRead, AsyncWrite};
 use ntex::connect::{self, Address, Connect, Connector};
-use ntex::framed::{ReadTask, State, WriteTask};
+use ntex::io::{Io, IoBoxed};
 use ntex::{service::Service, time::Seconds, util::ByteString, util::PoolId, util::PoolRef};
 
 #[cfg(feature = "openssl")]
@@ -41,8 +40,7 @@ where
 impl<A, T> RedisConnector<A, T>
 where
     A: Address + Clone,
-    T: Service<Request = Connect<A>, Error = connect::ConnectError>,
-    T::Response: AsyncRead + AsyncWrite + Unpin + 'static,
+    T: Service<Request = Connect<A>, Response = IoBoxed, Error = connect::ConnectError>,
 {
     /// Add redis auth password
     pub fn password<U>(mut self, password: U) -> Self
@@ -63,26 +61,10 @@ where
         self
     }
 
-    #[doc(hidden)]
-    #[deprecated(since = "0.2.4", note = "Use memory pool config")]
-    #[inline]
-    /// Set read/write buffer params
-    ///
-    /// By default read buffer is 16kb, write buffer is 16kb
-    pub fn buffer_params(
-        self,
-        _max_read_buf_size: u16,
-        _max_write_buf_size: u16,
-        _min_buf_size: u16,
-    ) -> Self {
-        self
-    }
-
     /// Use custom connector
-    pub fn connector<U>(self, connector: U) -> RedisConnector<A, U>
+    pub fn connector<U, F>(self, connector: U) -> RedisConnector<A, U>
     where
-        U: Service<Request = Connect<A>, Error = connect::ConnectError>,
-        U::Response: AsyncRead + AsyncWrite + Unpin + 'static,
+        U: Service<Request = Connect<A>, Response = Io<F>, Error = connect::ConnectError>,
     {
         RedisConnector {
             connector,
@@ -125,14 +107,10 @@ where
 
         async move {
             let io = fut.await?;
+            io.set_memory_pool(pool);
+            io.set_disconnect_timeout(Seconds::ZERO.into());
 
-            let state = State::with_memory_pool(pool);
-            state.set_disconnect_timeout(Seconds::ZERO);
-            let io = Rc::new(RefCell::new(io));
-            ntex::rt::spawn(ReadTask::new(io.clone(), state.clone()));
-            ntex::rt::spawn(WriteTask::new(io, state.clone()));
-
-            let client = Client::new(state);
+            let client = Client::new(io);
 
             if passwords.is_empty() {
                 Ok(client)
@@ -155,14 +133,10 @@ where
 
         async move {
             let io = fut.await?;
+            io.set_memory_pool(pool);
+            io.set_disconnect_timeout(Seconds::ZERO.into());
 
-            let state = State::with_memory_pool(pool);
-            state.set_disconnect_timeout(Seconds::ZERO);
-            let io = Rc::new(RefCell::new(io));
-            ntex::rt::spawn(ReadTask::new(io.clone(), state.clone()));
-            ntex::rt::spawn(WriteTask::new(io, state.clone()));
-
-            let mut client = SimpleClient::new(state);
+            let mut client = SimpleClient::new(io);
 
             if passwords.is_empty() {
                 Ok(client)

src/simple.rs

@@ -1,43 +1,42 @@
 use std::task::Poll;
 
-use ntex::{framed::State, util::poll_fn};
+use ntex::{io::IoBoxed, util::poll_fn};
 
 use super::cmd::Command;
 use super::codec::Codec;
 use super::errors::{CommandError, Error};
 
 /// Redis client
 pub struct SimpleClient {
-    state: State,
+    io: IoBoxed,
 }
 
 impl SimpleClient {
     /// Create new simple client
-    pub(crate) fn new(state: State) -> Self {
-        SimpleClient { state }
+    pub(crate) fn new(io: IoBoxed) -> Self {
+        SimpleClient { io }
     }
 
     /// Execute redis command
     pub async fn exec<U>(&mut self, cmd: U) -> Result<U::Output, CommandError>
     where
         U: Command,
     {
-        self.state.write().encode(cmd.to_request(), &Codec)?;
+        self.io.write().encode(cmd.to_request(), &Codec)?;
 
-        let read = self.state.read();
+        let read = self.io.read();
         poll_fn(|cx| {
             if let Some(item) = read.decode(&Codec)? {
                 return Poll::Ready(U::to_output(
                     item.into_result().map_err(CommandError::Error)?,
                 ));
             }
 
-            if !self.state.is_open() {
-                return Poll::Ready(Err(CommandError::Protocol(Error::Disconnected)));
+            if let Err(err) = read.poll_read_ready(cx) {
+                Poll::Ready(Err(CommandError::Protocol(Error::Io(err))))
+            } else {
+                Poll::Pending
             }
-
-            self.state.register_dispatcher(cx.waker());
-            Poll::Pending
         })
         .await
     }