memory pools support

Author: fafhrd91

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.2.4] - 2021-12-02
+
+* Add memory pools support
+
 ## [0.2.3] - 2021-09-03
 
 * Fix panic during protocol decoding

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-redis"
-version = "0.2.3"
+version = "0.2.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.4.0-b.11"
+ntex = "0.4.11"
 itoa = "0.4.5"
 btoi = "0.4.2"
 log = "0.4"

src/connector.rs

@@ -3,7 +3,7 @@ use std::{cell::RefCell, future::Future, rc::Rc};
 use ntex::codec::{AsyncRead, AsyncWrite};
 use ntex::connect::{self, Address, Connect, Connector};
 use ntex::framed::{ReadTask, State, WriteTask};
-use ntex::{service::Service, time::Seconds, util::ByteString};
+use ntex::{service::Service, time::Seconds, util::ByteString, util::PoolId, util::PoolRef};
 
 #[cfg(feature = "openssl")]
 use ntex::connect::openssl::{OpensslConnector, SslConnector};
@@ -19,9 +19,7 @@ pub struct RedisConnector<A, T> {
     address: A,
     connector: T,
     passwords: Vec<ByteString>,
-    lw: u16,
-    read_hw: u16,
-    write_hw: u16,
+    pool: PoolRef,
 }
 
 impl<A> RedisConnector<A, ()>
@@ -35,9 +33,7 @@ where
             address,
             passwords: Vec::new(),
             connector: Connector::default(),
-            lw: 1024,
-            read_hw: 16 * 1024,
-            write_hw: 16 * 1024,
+            pool: PoolId::P7.pool_ref(),
         }
     }
 }
@@ -58,19 +54,27 @@ where
         self
     }
 
+    /// Set memory pool.
+    ///
+    /// Use specified memory pool for memory allocations. By default P7
+    /// memory pool is used.
+    pub fn memory_pool(mut self, id: PoolId) -> Self {
+        self.pool = id.pool_ref();
+        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(
-        mut self,
-        max_read_buf_size: u16,
-        max_write_buf_size: u16,
-        min_buf_size: u16,
+        self,
+        _max_read_buf_size: u16,
+        _max_write_buf_size: u16,
+        _min_buf_size: u16,
     ) -> Self {
-        self.read_hw = max_read_buf_size;
-        self.write_hw = max_write_buf_size;
-        self.lw = min_buf_size;
         self
     }
 
@@ -84,9 +88,7 @@ where
             connector,
             address: self.address,
             passwords: self.passwords,
-            read_hw: self.read_hw,
-            write_hw: self.write_hw,
-            lw: self.lw,
+            pool: self.pool,
         }
     }
 
@@ -97,9 +99,7 @@ where
             address: self.address,
             passwords: self.passwords,
             connector: OpensslConnector::new(connector),
-            read_hw: self.read_hw,
-            write_hw: self.write_hw,
-            lw: self.lw,
+            pool: self.pool,
         }
     }
 
@@ -113,24 +113,21 @@ where
             address: self.address,
             passwords: self.passwords,
             connector: RustlsConnector::new(config),
-            read_hw: self.read_hw,
-            write_hw: self.write_hw,
-            lw: self.lw,
+            pool: self.pool,
         }
     }
 
     /// Connect to redis server and create shared client
     pub fn connect(&self) -> impl Future<Output = Result<Client, ConnectError>> {
-        let read_hw = self.read_hw;
-        let write_hw = self.write_hw;
-        let lw = self.lw;
+        let pool = self.pool;
         let passwords = self.passwords.clone();
         let fut = self.connector.call(Connect::new(self.address.clone()));
 
         async move {
             let io = fut.await?;
 
-            let state = State::with_params(read_hw, write_hw, lw, Seconds::ZERO);
+            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()));
@@ -152,16 +149,15 @@ where
 
     /// Connect to redis server and create simple client
     pub fn connect_simple(&self) -> impl Future<Output = Result<SimpleClient, ConnectError>> {
-        let read_hw = self.read_hw;
-        let write_hw = self.write_hw;
-        let lw = self.lw;
+        let pool = self.pool;
         let passwords = self.passwords.clone();
         let fut = self.connector.call(Connect::new(self.address.clone()));
 
         async move {
             let io = fut.await?;
 
-            let state = State::with_params(read_hw, write_hw, lw, Seconds::ZERO);
+            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()));