Use async fn in trait for Service definition

Author: fafhrd91

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.0.0-b.0] - 2024-01-07
+
+* Use "async fn" in trait for Service definition
+
 ## [0.5.1] - 2023-06-23
 
 * Fix client connector usage, fixes lifetime constraint

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-redis"
-version = "0.5.1"
+version = "1.0.0-b.0"
 authors = ["ntex contributors <[email protected]>"]
 description = "Redis client"
 documentation = "https://docs.rs/ntex-redis"
@@ -12,7 +12,7 @@ exclude = [".gitignore", ".travis.yml", ".cargo/config"]
 edition = "2018"
 
 [dependencies]
-ntex = "0.7.2"
+ntex = "1.0.0-b.0"
 itoa = "1.0.0"
 btoi = "0.4.2"
 log = "0.4"
@@ -21,4 +21,4 @@ derive_more = "0.99"
 [dev-dependencies]
 rand = "0.8"
 env_logger = "0.10"
-ntex = { version = "0.7.2", features = ["tokio"] }
+ntex = { version = "1.0.0-b.0", features = ["tokio"] }

src/client.rs

@@ -1,8 +1,8 @@
 use std::collections::VecDeque;
-use std::{cell::RefCell, fmt, future::Future, pin::Pin, rc::Rc, task::Context, task::Poll};
+use std::{cell::RefCell, fmt, future::poll_fn, rc::Rc, task::Context, task::Poll};
 
 use ntex::io::{IoBoxed, IoRef, OnDisconnect, RecvError};
-use ntex::util::{poll_fn, ready, Either, Ready};
+use ntex::util::ready;
 use ntex::{channel::pool, service::Service, service::ServiceCtx};
 
 use super::cmd::Command;
@@ -77,21 +77,17 @@ impl Client {
     }
 
     /// Execute redis command
-    pub fn exec<T>(&self, cmd: T) -> impl Future<Output = Result<T::Output, CommandError>>
+    pub async fn exec<T>(&self, cmd: T) -> Result<T::Output, CommandError>
     where
         T: Command,
     {
-        let is_open = !self.io.is_closed();
-        let fut = self._call(cmd.to_request());
-
-        async move {
-            if !is_open {
-                Err(CommandError::Protocol(Error::PeerGone(None)))
-            } else {
-                fut.await
-                    .map_err(CommandError::Protocol)
-                    .and_then(|res| T::to_output(res.into_result().map_err(CommandError::Error)?))
-            }
+        if self.io.is_closed() {
+            Err(CommandError::Protocol(Error::PeerGone(None)))
+        } else {
+            self._call(cmd.to_request())
+                .await
+                .map_err(CommandError::Protocol)
+                .and_then(|res| T::to_output(res.into_result().map_err(CommandError::Error)?))
         }
     }
 
@@ -106,21 +102,23 @@ impl Client {
         !self.io.is_closed()
     }
 
-    fn _call(&self, req: Request) -> Either<CommandResult, Ready<Response, Error>> {
+    async fn _call(&self, req: Request) -> Result<Response, Error> {
         if let Err(e) = self.io.encode(req, &Codec) {
-            Either::Right(Ready::Err(e))
+            Err(e)
         } else {
             let (tx, rx) = self.pool.channel();
             self.queue.borrow_mut().push_back(tx);
-            Either::Left(CommandResult { rx })
+            poll_fn(|cx| rx.poll_recv(cx))
+                .await
+                .map_err(|_| Error::PeerGone(None))
+                .and_then(|v| v)
         }
     }
 }
 
 impl Service<Request> for Client {
     type Response = Response;
     type Error = Error;
-    type Future<'f> = Either<CommandResult, Ready<Response, Error>>;
 
     fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         if self.disconnect.poll_ready(cx).is_ready() {
@@ -130,8 +128,8 @@ impl Service<Request> for Client {
         }
     }
 
-    fn call<'a>(&'a self, req: Request, _: ServiceCtx<'a, Self>) -> Self::Future<'a> {
-        self._call(req)
+    async fn call(&self, req: Request, _: ServiceCtx<'_, Self>) -> Result<Response, Error> {
+        self._call(req).await
     }
 }
 
@@ -142,18 +140,3 @@ impl fmt::Debug for Client {
             .finish()
     }
 }
-
-pub struct CommandResult {
-    rx: pool::Receiver<Result<Response, Error>>,
-}
-
-impl Future for CommandResult {
-    type Output = Result<Response, Error>;
-
-    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

@@ -75,11 +75,11 @@ where
     async fn _connect(&self) -> Result<IoBoxed, ConnectError> {
         let io: IoBoxed = self
             .connector
-            .service_call(Connect::new(self.address.clone()))
+            .call(Connect::new(self.address.clone()))
             .await?
             .into();
         io.set_memory_pool(self.pool);
-        io.set_disconnect_timeout(Seconds::ZERO.into());
+        io.set_disconnect_timeout(Seconds::ZERO);
 
         if self.passwords.is_empty() {
             Ok(io)

src/lib.rs

@@ -38,7 +38,7 @@ mod connector;
 pub mod errors;
 mod simple;
 
-pub use self::client::{Client, CommandResult};
+pub use self::client::Client;
 pub use self::connector::RedisConnector;
 pub use self::simple::{SimpleClient, SubscriptionClient};
 

src/simple.rs

@@ -1,13 +1,10 @@
 use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::{future::poll_fn, task::Context, task::Poll};
 
-use super::cmd::{
-    commands::{PubSubCommand, SubscribeOutputCommand},
-    Command,
-};
+use super::cmd::{commands::PubSubCommand, commands::SubscribeOutputCommand, Command};
 use super::codec::Codec;
 use super::errors::{CommandError, Error};
-use ntex::{io::IoBoxed, io::RecvError, util::poll_fn, util::ready, util::Stream};
+use ntex::{io::IoBoxed, io::RecvError, util::ready, util::Stream};
 
 /// Redis client
 pub struct SimpleClient {