Migrate to ntex 0.7

Author: fafhrd91

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.5.0] - 2023-06-22
+
+* Migrate to ntex 0.7
+
 ## [0.4.1] - 2023-01-28
 
 * Fix decode uncomple array data

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-redis"
-version = "0.4.1"
+version = "0.5.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.6.2"
+ntex = "0.7.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.6.2", features = ["tokio"] }
+ntex = { version = "0.7.0", features = ["tokio"] }

src/client.rs

@@ -3,7 +3,7 @@ use std::{cell::RefCell, fmt, future::Future, pin::Pin, rc::Rc, task::Context, t
 
 use ntex::io::{IoBoxed, IoRef, OnDisconnect, RecvError};
 use ntex::util::{poll_fn, ready, Either, Ready};
-use ntex::{channel::pool, service::Service};
+use ntex::{channel::pool, service::ServiceCtx, service::Service};
 
 use super::cmd::Command;
 use super::codec::{Codec, Request, Response};
@@ -82,7 +82,7 @@ impl Client {
         T: Command,
     {
         let is_open = !self.io.is_closed();
-        let fut = self.call(cmd.to_request());
+        let fut = self._call(cmd.to_request());
 
         async move {
             if !is_open {
@@ -97,14 +97,24 @@ impl Client {
 
     /// Delete all the keys of the currently selected DB.
     pub async fn flushdb(&self) -> Result<(), Error> {
-        self.call("FLUSHDB".into()).await?;
+        self._call("FLUSHDB".into()).await?;
         Ok(())
     }
 
     /// Returns true if underlying transport is connected to redis
     pub fn is_connected(&self) -> bool {
         !self.io.is_closed()
     }
+
+    fn _call(&self, req: Request) -> Either<CommandResult, Ready<Response, Error>> {
+        if let Err(e) = self.io.encode(req, &Codec) {
+            Either::Right(Ready::Err(e))
+        } else {
+            let (tx, rx) = self.pool.channel();
+            self.queue.borrow_mut().push_back(tx);
+            Either::Left(CommandResult { rx })
+        }
+    }
 }
 
 impl Service<Request> for Client {
@@ -120,14 +130,8 @@ impl Service<Request> for Client {
         }
     }
 
-    fn call(&self, req: Request) -> Self::Future<'_> {
-        if let Err(e) = self.io.encode(req, &Codec) {
-            Either::Right(Ready::Err(e))
-        } else {
-            let (tx, rx) = self.pool.channel();
-            self.queue.borrow_mut().push_back(tx);
-            Either::Left(CommandResult { rx })
-        }
+    fn call<'a>(&'a self, req: Request, _: ServiceCtx<'a, Self>) -> Self::Future<'a> {
+        self._call(req)
     }
 }
 

src/connector.rs

@@ -1,14 +1,13 @@
 use ntex::connect::{self, Address, Connect, Connector};
-use ntex::io::IoBoxed;
-use ntex::{service::Service, time::Seconds, util::ByteString, util::PoolId, util::PoolRef};
+use ntex::service::{Pipeline, Service};
+use ntex::{io::IoBoxed, time::Seconds, util::ByteString, util::PoolId, util::PoolRef};
 
-use super::errors::ConnectError;
-use super::{cmd, Client, SimpleClient};
+use super::{cmd, errors::ConnectError, Client, SimpleClient};
 
 /// Redis connector
 pub struct RedisConnector<A, T> {
     address: A,
-    connector: T,
+    connector: Pipeline<T>,
     passwords: Vec<ByteString>,
     pool: PoolRef,
 }
@@ -23,7 +22,7 @@ where
         RedisConnector {
             address,
             passwords: Vec::new(),
-            connector: Connector::default(),
+            connector: Pipeline::new(Connector::default()),
             pool: PoolId::P7.pool_ref(),
         }
     }
@@ -59,7 +58,7 @@ where
         IoBoxed: From<U::Response>,
     {
         RedisConnector {
-            connector,
+            connector: Pipeline::new(connector),
             address: self.address,
             passwords: self.passwords,
             pool: self.pool,