update to 0.5.0-b.1

Author: fafhrd91

Cargo.toml

@@ -21,7 +21,7 @@ openssl = ["ntex/openssl"]
 rustls = ["ntex/rustls"]
 
 [dependencies]
-ntex = "0.5.0-b.0"
+ntex = "0.5.0-b.1"
 itoa = "0.4.5"
 btoi = "0.4.2"
 log = "0.4"

src/client.rs

@@ -2,7 +2,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::util::{poll_fn, Either, Ready};
+use ntex::util::{poll_fn, ready, Either, Ready};
 use ntex::{channel::pool, service::Service};
 
 use super::cmd::Command;
@@ -28,38 +28,32 @@ impl Client {
         let io_ref = io.get_ref();
         let queue2 = queue.clone();
         ntex::rt::spawn(async move {
-            let read = io.read();
-
-            poll_fn(|cx| {
-                loop {
-                    match read.decode(&Codec) {
-                        Err(e) => {
-                            if let Some(tx) = queue2.borrow_mut().pop_front() {
-                                let _ = tx.send(Err(e));
-                            }
-                            queue2.borrow_mut().clear();
-                            let _ = io.poll_shutdown(cx);
-                            return Poll::Ready(());
-                        }
-                        Ok(Some(item)) => {
-                            if let Some(tx) = queue2.borrow_mut().pop_front() {
-                                let _ = tx.send(Ok(item));
-                            } else {
-                                log::error!("Unexpected redis response: {:?}", item);
-                            }
-                        }
-                        Ok(None) => break,
+            poll_fn(|cx| match ready!(io.poll_read_next(&Codec, cx)) {
+                Some(Ok(item)) => {
+                    if let Some(tx) = queue2.borrow_mut().pop_front() {
+                        let _ = tx.send(Ok(item));
+                    } else {
+                        log::error!("Unexpected redis response: {:?}", item);
                     }
+                    Poll::Pending
                 }
-
-                if io.is_closed() {
+                Some(Err(Either::Left(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(());
                 }
-                if read.poll_read_ready(cx).is_err() {
-                    Poll::Ready(())
-                } else {
-                    Poll::Pending
+                Some(Err(Either::Right(e))) => {
+                    if let Some(tx) = queue2.borrow_mut().pop_front() {
+                        let _ = tx.send(Err(e.into()));
+                    }
+                    queue2.borrow_mut().clear();
+                    let _ = ready!(io.poll_shutdown(cx));
+                    return Poll::Ready(());
                 }
+                None => Poll::Ready(()),
             })
             .await
         });
@@ -120,7 +114,7 @@ impl Service for Client {
     }
 
     fn call(&self, req: Request) -> Self::Future {
-        if let Err(e) = self.io.write().encode(req, &Codec) {
+        if let Err(e) = self.io.encode(req, &Codec) {
             Either::Right(Ready::Err(e))
         } else {
             let (tx, rx) = self.pool.channel();

src/connector.rs

@@ -157,7 +157,7 @@ where
             io.set_memory_pool(pool);
             io.set_disconnect_timeout(Seconds::ZERO.into());
 
-            let mut client = SimpleClient::new(io);
+            let client = SimpleClient::new(io);
 
             if passwords.is_empty() {
                 Ok(client)

src/simple.rs

@@ -1,6 +1,6 @@
 use std::task::Poll;
 
-use ntex::{io::IoBoxed, util::poll_fn};
+use ntex::{io::IoBoxed, util::poll_fn, util::ready, util::Either};
 
 use super::cmd::Command;
 use super::codec::Codec;
@@ -18,25 +18,19 @@ impl SimpleClient {
     }
 
     /// Execute redis command
-    pub async fn exec<U>(&mut self, cmd: U) -> Result<U::Output, CommandError>
+    pub async fn exec<U>(&self, cmd: U) -> Result<U::Output, CommandError>
     where
         U: Command,
     {
-        self.io.write().encode(cmd.to_request(), &Codec)?;
+        self.io.encode(cmd.to_request(), &Codec)?;
 
-        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 let Err(err) = read.poll_read_ready(cx) {
-                Poll::Ready(Err(CommandError::Protocol(Error::Io(err))))
-            } else {
-                Poll::Pending
-            }
+        poll_fn(|cx| match ready!(self.io.poll_read_next(&Codec, cx)) {
+            Some(Ok(item)) => Poll::Ready(U::to_output(
+                item.into_result().map_err(CommandError::Error)?,
+            )),
+            Some(Err(Either::Left(err))) => Poll::Ready(Err(CommandError::Protocol(err))),
+            Some(Err(Either::Right(err))) => Poll::Ready(Err(CommandError::Protocol(err.into()))),
+            None => Poll::Ready(Err(CommandError::Protocol(Error::Disconnected))),
         })
         .await
     }