upgrade to ntex 0.5-b.4

Author: fafhrd91

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-redis"
-version = "0.3.0-b.3"
+version = "0.3.0-b.4"
 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.5.0-b.3"
+ntex = "0.5.0-b.4"
 itoa = "0.4.5"
 btoi = "0.4.2"
 log = "0.4"

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::{IoBoxed, IoRef, OnDisconnect};
+use ntex::io::{IoBoxed, IoRef, OnDisconnect, RecvError};
 use ntex::util::{poll_fn, ready, Either, Ready};
 use ntex::{channel::pool, service::Service};
 
@@ -30,31 +30,37 @@ impl Client {
         ntex::rt::spawn(async move {
             poll_fn(|cx| loop {
                 match ready!(io.poll_recv(&Codec, cx)) {
-                    Ok(Some(item)) => {
+                    Ok(item) => {
                         if let Some(tx) = queue2.borrow_mut().pop_front() {
                             let _ = tx.send(Ok(item));
                         } else {
                             log::error!("Unexpected redis response: {:?}", item);
                         }
                         continue;
                     }
-                    Err(Either::Left(e)) => {
+                    Err(RecvError::KeepAlive) | Err(RecvError::Stop) => {
+                        unreachable!()
+                    }
+                    Err(RecvError::WriteBackpressure) => {
+                        if ready!(io.poll_flush(cx, false)).is_err() {
+                            return Poll::Ready(());
+                        } else {
+                            continue;
+                        }
+                    }
+                    Err(RecvError::Decoder(e)) => {
                         if let Some(tx) = queue2.borrow_mut().pop_front() {
                             let _ = tx.send(Err(e));
                         }
                         queue2.borrow_mut().clear();
                         let _ = ready!(io.poll_shutdown(cx));
                         return Poll::Ready(());
                     }
-                    Err(Either::Right(e)) => {
-                        if let Some(tx) = queue2.borrow_mut().pop_front() {
-                            let _ = tx.send(Err(e.into()));
-                        }
+                    Err(RecvError::PeerGone(e)) => {
+                        log::info!("Redis connection is dropped: {:?}", e);
                         queue2.borrow_mut().clear();
-                        let _ = ready!(io.poll_shutdown(cx));
                         return Poll::Ready(());
                     }
-                    Ok(None) => return Poll::Ready(()),
                 }
             })
             .await
@@ -80,7 +86,7 @@ impl Client {
 
         async move {
             if !is_open {
-                Err(CommandError::Protocol(Error::Disconnected))
+                Err(CommandError::Protocol(Error::PeerGone(None)))
             } else {
                 fut.await
                     .map_err(CommandError::Protocol)
@@ -108,7 +114,7 @@ impl Service<Request> for Client {
 
     fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         if self.disconnect.poll_ready(cx).is_ready() {
-            Poll::Ready(Err(Error::Disconnected))
+            Poll::Ready(Err(Error::PeerGone(None)))
         } else {
             Poll::Ready(Ok(()))
         }
@@ -143,7 +149,7 @@ impl Future for CommandResult {
     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::Disconnected)),
+            Poll::Ready(Err(_)) => Poll::Ready(Err(Error::PeerGone(None))),
             Poll::Pending => Poll::Pending,
         }
     }

src/errors.rs

@@ -15,11 +15,7 @@ pub enum Error {
 
     /// An IO error occurred
     #[display(fmt = "Io error: {:?}", _0)]
-    Io(Option<io::Error>),
-
-    /// Connection is disconnected
-    #[display(fmt = "Redis server has been disconnected")]
-    Disconnected,
+    PeerGone(Option<io::Error>),
 }
 
 impl std::error::Error for Error {}
@@ -28,23 +24,22 @@ impl Clone for Error {
     fn clone(&self) -> Self {
         match self {
             Error::Parse(_) => Error::Parse(String::new()),
-            Error::Io(_) => Error::Io(None),
-            Error::Disconnected => Error::Disconnected,
+            Error::PeerGone(_) => Error::PeerGone(None),
         }
     }
 }
 
 impl From<io::Error> for Error {
     fn from(err: io::Error) -> Error {
-        Error::Io(Some(err))
+        Error::PeerGone(Some(err))
     }
 }
 
 impl From<Either<Error, io::Error>> for Error {
     fn from(err: Either<Error, io::Error>) -> Error {
         match err {
             Either::Left(err) => err,
-            Either::Right(err) => Error::Io(Some(err)),
+            Either::Right(err) => Error::PeerGone(Some(err)),
         }
     }
 }

src/simple.rs

@@ -1,6 +1,6 @@
 use std::task::Poll;
 
-use ntex::{io::IoBoxed, util::poll_fn, util::ready, util::Either};
+use ntex::{io::IoBoxed, io::RecvError, util::poll_fn, util::ready};
 
 use super::cmd::Command;
 use super::codec::Codec;
@@ -24,13 +24,24 @@ impl SimpleClient {
     {
         self.io.encode(cmd.to_request(), &Codec)?;
 
-        poll_fn(|cx| match ready!(self.io.poll_recv(&Codec, cx)) {
-            Ok(Some(item)) => Poll::Ready(U::to_output(
-                item.into_result().map_err(CommandError::Error)?,
-            )),
-            Err(Either::Left(err)) => Poll::Ready(Err(CommandError::Protocol(err))),
-            Err(Either::Right(err)) => Poll::Ready(Err(CommandError::Protocol(err.into()))),
-            Ok(None) => Poll::Ready(Err(CommandError::Protocol(Error::Disconnected))),
+        poll_fn(|cx| loop {
+            return match ready!(self.io.poll_recv(&Codec, cx)) {
+                Ok(item) => Poll::Ready(U::to_output(
+                    item.into_result().map_err(CommandError::Error)?,
+                )),
+                Err(RecvError::KeepAlive) | Err(RecvError::Stop) => {
+                    unreachable!()
+                }
+                Err(RecvError::WriteBackpressure) => {
+                    ready!(self.io.poll_flush(cx, false))
+                        .map_err(|e| CommandError::Protocol(Error::PeerGone(Some(e))))?;
+                    continue;
+                }
+                Err(RecvError::Decoder(err)) => Poll::Ready(Err(CommandError::Protocol(err))),
+                Err(RecvError::PeerGone(err)) => {
+                    Poll::Ready(Err(CommandError::Protocol(Error::PeerGone(err))))
+                }
+            };
         })
         .await
     }