Do not check readiness for call (ntex-rs#178)

Author: fafhrd91

CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [4.1.0] - 2024-10-10
+
+* Do not check readiness for call
+
+* Handle service readiness errors during shutdown
+
 ## [4.0.0] - 2024-10-05
 
 * Middlewares support for mqtt server

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-mqtt"
-version = "4.0.0"
+version = "4.1.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Client and Server framework for MQTT v5 and v3.1.1 protocols"
 documentation = "https://docs.rs/ntex-mqtt"

src/io.rs

@@ -2,7 +2,6 @@
 use std::task::{ready, Context, Poll};
 use std::{cell::RefCell, collections::VecDeque, future::Future, pin::Pin, rc::Rc};
 
-use ntex_bytes::Pool;
 use ntex_codec::{Decoder, Encoder};
 use ntex_io::{
     Decoded, DispatchItem, DispatcherConfig, IoBoxed, IoRef, IoStatusUpdate, RecvError,
@@ -22,8 +21,7 @@ pin_project_lite::pin_project! {
         U: Decoder,
         U: 'static,
     {
-        inner: DispatcherInner<S, U>,
-        pool: Pool,
+        inner: DispatcherInner<S, U>
     }
 }
 
@@ -122,11 +120,9 @@ where
             base: 0,
             queue: VecDeque::new(),
         }));
-        let pool = io.memory_pool().pool();
         let keepalive_timeout = config.keepalive_timeout();
 
         Dispatcher {
-            pool,
             inner: DispatcherInner {
                 io,
                 codec,
@@ -249,14 +245,6 @@ where
             }
         }
 
-        // handle memory pool pressure
-        if this.pool.poll_ready(cx).is_pending() {
-            inner.flags.remove(Flags::KA_TIMEOUT | Flags::READ_TIMEOUT);
-            inner.io.stop_timer();
-            inner.io.pause();
-            return Poll::Pending;
-        }
-
         loop {
             match inner.st {
                 IoDispatcherState::Processing => {
@@ -312,9 +300,7 @@ where
                 IoDispatcherState::Backpressure => {
                     match ready!(inner.poll_service(cx)) {
                         PollService::Ready => (),
-                        PollService::Item(item) => {
-                            inner.call_service(cx, item);
-                        }
+                        PollService::Item(item) => inner.call_service(cx, item),
                         PollService::Continue => continue,
                     };
 
@@ -334,7 +320,11 @@ where
 
                     // service may relay on poll_ready for response results
                     if !inner.flags.contains(Flags::READY_ERR) {
-                        let _ = inner.service.poll_ready(cx);
+                        if let Poll::Ready(res) = inner.service.poll_ready(cx) {
+                            if res.is_err() {
+                                inner.flags.insert(Flags::READY_ERR);
+                            }
+                        }
                     }
 
                     if inner.state.borrow().queue.is_empty() {
@@ -394,7 +384,7 @@ where
 {
     fn call_service(&mut self, cx: &mut Context<'_>, item: DispatchItem<U>) {
         let mut state = self.state.borrow_mut();
-        let mut fut = self.service.call(item);
+        let mut fut = self.service.call_nowait(item);
 
         // optimize first call
         if self.response.is_none() {
@@ -429,8 +419,7 @@ where
             let codec = self.codec.clone();
             let state = self.state.clone();
 
-            #[allow(clippy::let_underscore_future)]
-            let _ = ntex_util::spawn(async move {
+            ntex_util::spawn(async move {
                 let item = fut.await;
                 state.borrow_mut().handle_result(item, response_idx, &st, &codec, true);
             });
@@ -502,8 +491,8 @@ where
             Poll::Ready(Err(err)) => {
                 log::trace!("{}: Service readiness check failed, stopping", self.io.tag());
                 self.st = IoDispatcherState::Stop;
-                self.state.borrow_mut().error = Some(IoDispatcherError::Service(err));
                 self.flags.insert(Flags::READY_ERR);
+                self.state.borrow_mut().error = Some(IoDispatcherError::Service(err));
                 Poll::Ready(PollService::Continue)
             }
         }
@@ -629,7 +618,6 @@ mod tests {
 
             (
                 Dispatcher {
-                    pool: io.memory_pool().pool(),
                     inner: DispatcherInner {
                         codec,
                         state,