.github/workflows/checks.yml

@@ -12,7 +12,7 @@ jobs:
         with:
           toolchain: stable
       - run:
-          cargo check --tests --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
+          cargo check --tests --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
 
   clippy:
     name: Clippy
@@ -24,7 +24,7 @@ jobs:
           toolchain: stable
           components: clippy
       - run:
-          cargo clippy --tests --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
+          cargo clippy --tests --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
 
   fmt:
     name: Rustfmt

.github/workflows/cov.yml

@@ -8,6 +8,11 @@ jobs:
     env:
       CARGO_TERM_COLOR: always
     steps:
+      - name: Free Disk Space
+        uses: jlumbroso/free-disk-space@main
+        with:
+          tool-cache: true
+
       - uses: actions/checkout@v4
       - name: Install Rust
         run: rustup update nightly
@@ -21,17 +26,17 @@ jobs:
       - name: Clean coverage results
         run: cargo llvm-cov clean --workspace
 
-      - name: Code coverage (neon)
-        run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
-
-      - name: Code coverage (neon-uring)
-        run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/neon-uring,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
-
       - name: Code coverage (tokio)
-        run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
+        run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
 
       - name: Code coverage (compio)
-        run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
+        run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
+
+      - name: Code coverage (neon)
+        run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
+
+      - name: Code coverage (neon-uring)
+        run: cargo llvm-cov --no-report --all --no-default-features --features="ntex/neon-uring,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
 
       - name: Generate coverage report
         run: cargo llvm-cov report --lcov --output-path lcov.info --ignore-filename-regex="ntex-compio|ntex-tokio"

.github/workflows/linux.yml

@@ -16,6 +16,11 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
+      - name: Free Disk Space
+        uses: jlumbroso/free-disk-space@main
+        with:
+          tool-cache: true
+
       - uses: actions/checkout@v4
 
       - name: Install ${{ matrix.version }}
@@ -39,25 +44,25 @@ jobs:
           path: ~/.cargo/git
           key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-index-trimmed-${{ hashFiles('**/Cargo.lock') }}
 
-      - name: Run tests (neon)
-        timeout-minutes: 40
-        run: |
-          cargo test --all --no-default-features --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
-
-      - name: Run tests (neon-uring)
-        timeout-minutes: 40
-        run: |
-          cargo test --all --no-default-features --features="ntex/neon-uring,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
-
       - name: Run tests (tokio)
         timeout-minutes: 40
         run: |
-          cargo test --all --no-fail-fast --no-default-features --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
+          cargo test --all --no-fail-fast --no-default-features --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
 
       - name: Run tests (compio)
         timeout-minutes: 40
         run: |
-          cargo test --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
+          cargo test --all --no-default-features --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
+
+      - name: Run tests (neon)
+        timeout-minutes: 40
+        run: |
+          cargo test --all --no-default-features --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
+
+      - name: Run tests (neon-uring)
+        timeout-minutes: 40
+        run: |
+          cargo test --all --no-default-features --features="ntex/neon-uring,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
 
       - name: Install cargo-cache
         continue-on-error: true

.github/workflows/osx.yml

@@ -37,16 +37,16 @@ jobs:
           path: ~/.cargo/git
           key: ${{ matrix.version }}-aarch64-apple-darwin-cargo-index-trimmed-${{ hashFiles('**/Cargo.lock') }}
 
-      - name: Run tests (neon)
-        timeout-minutes: 40
-        run: cargo test --all --no-default-features --no-fail-fast --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
-
       - name: Run tests (tokio)
-        run: cargo test --all --no-default-features --no-fail-fast --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
+        run: cargo test --all --no-default-features --no-fail-fast --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
 
       - name: Run tests (compio)
         timeout-minutes: 40
-        run: cargo test --all --no-default-features --no-fail-fast --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws"
+        run: cargo test --all --no-default-features --no-fail-fast --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
+
+      - name: Run tests (neon)
+        timeout-minutes: 40
+        run: cargo test --all --no-default-features --no-fail-fast --features="ntex/neon,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli"
 
       - name: Install cargo-cache
         continue-on-error: true

.github/workflows/windows.yml

@@ -63,8 +63,8 @@ jobs:
 
       - name: Run tests (tokio)
         run: |
-          cargo test --all --lib --no-default-features --no-fail-fast --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" -- --skip test_timer
+          cargo test --all --lib --no-default-features --no-fail-fast --features="ntex/tokio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" -- --skip test_timer
 
       - name: Run tests (compio)
         run: |
-          cargo test --all --lib --no-default-features --no-fail-fast --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws" -- --skip test_timer
+          cargo test --all --lib --no-default-features --no-fail-fast --features="ntex/compio,ntex/cookie,ntex/url,ntex/compress,ntex/openssl,ntex/rustls,ntex/ws,ntex/brotli" -- --skip test_timer

Cargo.toml

@@ -46,12 +46,7 @@ ntex-compio = { path = "ntex-compio" }
 ntex-tokio = { path = "ntex-tokio" }
 
 [workspace.dependencies]
-ntex-polling = "3.7.4"
-
-async-channel = "2"
 async-task = "4.5.0"
-atomic-waker = "1.1"
-core_affinity = "0.8"
 bitflags = "2"
 cfg_aliases = "0.2.1"
 cfg-if = "1.0.0"
@@ -62,8 +57,7 @@ fxhash = "0.2"
 libc = "0.2.164"
 log = "0.4"
 io-uring = "0.7.4"
-oneshot = "0.1"
-polling = "3.7.4"
+polling = "3.3.0"
 nohash-hasher = "0.2.0"
 scoped-tls = "1.0.1"
 slab = "0.4.9"

ntex-bytes/src/bytes.rs

@@ -965,7 +965,7 @@ impl From<String> for Bytes {
     fn from(src: String) -> Bytes {
         if src.is_empty() {
             Bytes::new()
-        } else if src.len() <= INLINE_CAP {
+        } else if src.bytes().len() <= INLINE_CAP {
             Bytes {
                 inner: Inner::from_slice_inline(src.as_bytes()),
             }

ntex-bytes/tests/test_bytes.rs

@@ -104,7 +104,6 @@ fn clone_mut() {
     assert_eq!(buf1, "hello");
 }
 
-#[allow(clippy::sliced_string_as_bytes)]
 #[test]
 fn fmt_write() {
     use std::fmt::Write;

ntex-io/CHANGES.md

@@ -1,9 +1,5 @@
 # Changes
 
-## [2.11.1] - 2025-03-20
-
-* Add readiness check support
-
 ## [2.11.0] - 2025-03-10
 
 * Add single io context

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "2.11.2"
+version = "2.11.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for encoding and decoding frames"
 keywords = ["network", "framework", "async", "futures"]
@@ -28,3 +28,4 @@ pin-project-lite = "0.2"
 [dev-dependencies]
 ntex = "2"
 rand = "0.8"
+env_logger = "0.11"

ntex-io/src/dispatcher.rs

@@ -543,7 +543,7 @@ where
             if self.flags.contains(Flags::KA_ENABLED)
                 && !self.flags.contains(Flags::KA_TIMEOUT)
             {
-                log::trace!(
+                log::debug!(
                     "{}: Start keep-alive timer {:?}",
                     self.shared.io.tag(),
                     self.cfg.keepalive_timeout()

ntex-io/src/io.rs

@@ -98,19 +98,17 @@ impl IoState {
     }
 
     pub(super) fn io_stopped(&self, err: Option<io::Error>) {
-        if !self.flags.get().contains(Flags::IO_STOPPED) {
-            if err.is_some() {
-                self.error.set(err);
-            }
-            self.read_task.wake();
-            self.write_task.wake();
-            self.dispatch_task.wake();
-            self.notify_disconnect();
-            self.handle.take();
-            self.insert_flags(
-                Flags::IO_STOPPED | Flags::IO_STOPPING | Flags::IO_STOPPING_FILTERS,
-            );
+        if err.is_some() {
+            self.error.set(err);
         }
+        self.read_task.wake();
+        self.write_task.wake();
+        self.dispatch_task.wake();
+        self.notify_disconnect();
+        self.handle.take();
+        self.insert_flags(
+            Flags::IO_STOPPED | Flags::IO_STOPPING | Flags::IO_STOPPING_FILTERS,
+        );
     }
 
     /// Gracefully shutdown read and write io tasks
@@ -535,8 +533,8 @@ impl<F> Io<F> {
             } else {
                 match self.poll_read_ready(cx) {
                     Poll::Pending | Poll::Ready(Ok(Some(()))) => {
-                        if log::log_enabled!(log::Level::Trace) && decoded.remains != 0 {
-                            log::trace!(
+                        if log::log_enabled!(log::Level::Debug) && decoded.remains != 0 {
+                            log::debug!(
                                 "{}: Not enough data to decode next frame",
                                 self.tag()
                             );

ntex-io/src/ioref.rs

@@ -246,12 +246,12 @@ impl IoRef {
             if cur_hnd.is_set() {
                 let hnd = timer::update(cur_hnd, timeout, self);
                 if hnd != cur_hnd {
-                    log::trace!("{}: Update timer {:?}", self.tag(), timeout);
+                    log::debug!("{}: Update timer {:?}", self.tag(), timeout);
                     self.0.timeout.set(hnd);
                 }
                 hnd
             } else {
-                log::trace!("{}: Start timer {:?}", self.tag(), timeout);
+                log::debug!("{}: Start timer {:?}", self.tag(), timeout);
                 let hnd = timer::register(timeout, self);
                 self.0.timeout.set(hnd);
                 hnd
@@ -270,7 +270,7 @@ impl IoRef {
     pub fn stop_timer(&self) {
         let hnd = self.0.timeout.get();
         if hnd.is_set() {
-            log::trace!("{}: Stop timer", self.tag());
+            log::debug!("{}: Stop timer", self.tag());
             self.0.timeout.set(timer::TimerHandle::ZERO);
             timer::unregister(hnd, self)
         }

ntex-io/src/tasks.rs

@@ -395,12 +395,6 @@ impl IoContext {
         self.0 .0.io_stopped(e);
     }
 
-    #[inline]
-    /// Check if Io stopped
-    pub fn is_stopped(&self) -> bool {
-        self.0.flags().is_stopped()
-    }
-
     /// Wait when io get closed or preparing for close
     pub async fn shutdown(&self, flush_buf: bool) {
         let st = &self.0 .0;
@@ -674,6 +668,7 @@ impl IoContext {
         // set buffer back
         let result = match result {
             Ok(0) => {
+                // log::debug!("{}: WROTE ALL {:?}", self.0.tag(), inner.buffer.write_destination_size());
                 self.0.memory_pool().release_write_buf(buf);
                 Ok(inner.buffer.write_destination_size())
             }
@@ -683,6 +678,7 @@ impl IoContext {
                     self.0.memory_pool().release_write_buf(b);
                 }
                 let l = buf.len();
+                // log::debug!("{}: WROTE SOME {:?}", self.0.tag(), l);
                 inner.buffer.set_write_destination(buf);
                 Ok(l)
             }
@@ -726,44 +722,81 @@ impl IoContext {
     }
 
     /// Get read buffer
-    pub fn is_read_ready(&self) -> bool {
-        // check read readiness
-        if let Some(waker) = self.0 .0.read_task.take() {
-            let mut cx = Context::from_waker(&waker);
-
-            if let Poll::Ready(ReadStatus::Ready) = self.0.filter().poll_read_ready(&mut cx)
-            {
-                return true;
-            }
-        }
-        false
-    }
-
     pub fn with_read_buf<F>(&self, f: F) -> Poll<()>
     where
-        F: FnOnce(&mut BytesVec, usize, usize) -> (usize, Poll<io::Result<()>>),
+        F: FnOnce(&mut BytesVec) -> Poll<io::Result<usize>>,
+    {
+        let result = self.with_read_buf_inner(f);
+
+        // check read readiness
+        if result.is_pending() {
+            if let Some(waker) = self.0 .0.read_task.take() {
+                let mut cx = Context::from_waker(&waker);
+
+                if let Poll::Ready(ReadStatus::Ready) =
+                    self.0.filter().poll_read_ready(&mut cx)
+                {
+                    return Poll::Pending;
+                }
+            }
+        }
+        result
+    }
+
+    fn with_read_buf_inner<F>(&self, f: F) -> Poll<()>
+    where
+        F: FnOnce(&mut BytesVec) -> Poll<io::Result<usize>>,
     {
         let inner = &self.0 .0;
         let (hw, lw) = self.0.memory_pool().read_params().unpack();
-        let (nbytes, result) = inner.buffer.with_read_source(&self.0, |buf| f(buf, hw, lw));
+        let result = inner.buffer.with_read_source(&self.0, |buf| {
+            // make sure we've got room
+            let remaining = buf.remaining_mut();
+            if remaining < lw {
+                buf.reserve(hw - remaining);
+            }
+
+            f(buf)
+        });
 
         // handle buffer changes
-        let st_res = if nbytes > 0 {
-            let filter = self.0.filter();
-            match filter.process_read_buf(&self.0, &inner.buffer, 0, nbytes) {
-                Ok(status) => {
-                    if status.nbytes > 0 {
-                        // dest buffer has new data, wake up dispatcher
-                        if inner.buffer.read_destination_size() >= hw {
+        match result {
+            Poll::Ready(Ok(0)) => {
+                inner.io_stopped(None);
+                Poll::Ready(())
+            }
+            Poll::Ready(Ok(nbytes)) => {
+                let filter = self.0.filter();
+                let _ = filter
+                    .process_read_buf(&self.0, &inner.buffer, 0, nbytes)
+                    .and_then(|status| {
+                        if status.nbytes > 0 {
+                            // dest buffer has new data, wake up dispatcher
+                            if inner.buffer.read_destination_size() >= hw {
+                                log::trace!(
+                                    "{}: Io read buffer is too large {}, enable read back-pressure",
+                                    self.0.tag(),
+                                    nbytes
+                                );
+                                inner.insert_flags(Flags::BUF_R_READY | Flags::BUF_R_FULL);
+                            } else {
+                                inner.insert_flags(Flags::BUF_R_READY);
+
+                                if nbytes >= hw {
+                                    // read task is paused because of read back-pressure
+                                    // but there is no new data in top most read buffer
+                                    // so we need to wake up read task to read more data
+                                    // otherwise read task would sleep forever
+                                    inner.read_task.wake();
+                                }
+                            }
                             log::trace!(
-                                "{}: Io read buffer is too large {}, enable read back-pressure",
+                                "{}: New {} bytes available, wakeup dispatcher",
                                 self.0.tag(),
                                 nbytes
                             );
-                            inner.insert_flags(Flags::BUF_R_READY | Flags::BUF_R_FULL);
+                            inner.dispatch_task.wake();
                         } else {
-                            inner.insert_flags(Flags::BUF_R_READY);
-
                             if nbytes >= hw {
                                 // read task is paused because of read back-pressure
                                 // but there is no new data in top most read buffer
@@ -771,76 +804,67 @@ impl IoContext {
                                 // otherwise read task would sleep forever
                                 inner.read_task.wake();
                             }
+                            if inner.flags.get().is_waiting_for_read() {
+                                // in case of "notify" we must wake up dispatch task
+                                // if we read any data from source
+                                inner.dispatch_task.wake();
+                            }
                         }
-                        log::trace!(
-                            "{}: New {} bytes available, wakeup dispatcher",
-                            self.0.tag(),
-                            nbytes
-                        );
+
+                        // while reading, filter wrote some data
+                        // in that case filters need to process write buffers
+                        // and potentialy wake write task
+                        if status.need_write {
+                            filter.process_write_buf(&self.0, &inner.buffer, 0)
+                        } else {
+                            Ok(())
+                        }
+                    })
+                    .map_err(|err| {
                         inner.dispatch_task.wake();
-                    } else {
-                        if nbytes >= hw {
-                            // read task is paused because of read back-pressure
-                            // but there is no new data in top most read buffer
-                            // so we need to wake up read task to read more data
-                            // otherwise read task would sleep forever
-                            inner.read_task.wake();
-                        }
-                        if inner.flags.get().is_waiting_for_read() {
-                            // in case of "notify" we must wake up dispatch task
-                            // if we read any data from source
-                            inner.dispatch_task.wake();
-                        }
-                    }
-
-                    // while reading, filter wrote some data
-                    // in that case filters need to process write buffers
-                    // and potentialy wake write task
-                    if status.need_write {
-                        filter.process_write_buf(&self.0, &inner.buffer, 0)
-                    } else {
-                        Ok(())
-                    }
-                }
-                Err(err) => {
-                    inner.insert_flags(Flags::BUF_R_READY);
-                    Err(err)
-                }
-            }
-        } else {
-            Ok(())
-        };
-
-        match result {
-            Poll::Ready(Ok(_)) => {
-                if let Err(e) = st_res {
-                    inner.io_stopped(Some(e));
-                    Poll::Ready(())
-                } else if nbytes == 0 {
-                    inner.io_stopped(None);
-                    Poll::Ready(())
-                } else {
-                    Poll::Pending
-                }
+                        inner.io_stopped(Some(err));
+                        inner.insert_flags(Flags::BUF_R_READY);
+                    });
+                Poll::Pending
             }
             Poll::Ready(Err(e)) => {
                 inner.io_stopped(Some(e));
                 Poll::Ready(())
             }
             Poll::Pending => {
-                if let Err(e) = st_res {
-                    inner.io_stopped(Some(e));
-                    Poll::Ready(())
-                } else {
-                    self.shutdown_filters();
-                    Poll::Pending
-                }
+                self.shutdown_filters();
+                Poll::Pending
             }
         }
     }
 
-    /// Get write buffer
     pub fn with_write_buf<F>(&self, f: F) -> Poll<()>
+    where
+        F: FnOnce(&BytesVec) -> Poll<io::Result<usize>>,
+    {
+        let result = self.with_write_buf_inner(f);
+
+        // check write readiness
+        if result.is_pending() {
+            let inner = &self.0 .0;
+            if let Some(waker) = inner.write_task.take() {
+                let ready = self
+                    .0
+                    .filter()
+                    .poll_write_ready(&mut Context::from_waker(&waker));
+                if !matches!(
+                    ready,
+                    Poll::Ready(WriteStatus::Ready | WriteStatus::Shutdown)
+                ) {
+                    return Poll::Ready(());
+                }
+            }
+        }
+        result
+    }
+
+    /// Get write buffer
+    fn with_write_buf_inner<F>(&self, f: F) -> Poll<()>
     where
         F: FnOnce(&BytesVec) -> Poll<io::Result<usize>>,
     {

ntex-macros/Cargo.toml

@@ -18,3 +18,4 @@ proc-macro2 = "^1"
 [dev-dependencies]
 ntex = "2"
 futures = "0.3"
+env_logger = "0.11"

ntex-net/CHANGES.md

@@ -1,43 +1,5 @@
 # Changes
 
-## [2.5.11] - 2025-04-05
-
-* Various improvements for polling driver
-
-## [2.5.10] - 2025-03-28
-
-* Better closed sockets handling
-
-## [2.5.9] - 2025-03-27
-
-* Handle closed sockets
-
-## [2.5.8] - 2025-03-25
-
-* Update neon runtime
-
-## [2.5.7] - 2025-03-21
-
-* Simplify neon poll impl
-
-## [2.5.6] - 2025-03-20
-
-* Redesign neon poll support
-
-## [2.5.5] - 2025-03-17
-
-* Add check for required io-uring opcodes
-
-* Handle io-uring cancelation
-
-## [2.5.4] - 2025-03-15
-
-* Close FD in various case for poll driver
-
-## [2.5.3] - 2025-03-14
-
-* Fix operation cancelation handling for poll driver
-
 ## [2.5.2] - 2025-03-14
 
 * Fix operation cancelation handling for io-uring driver

ntex-net/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-net"
-version = "2.5.11"
+version = "2.5.2"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntexwork utils for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -27,32 +27,34 @@ compio = ["ntex-rt/compio", "ntex-compio"]
 # neon runtime
 neon = ["ntex-rt/neon", "ntex-neon", "slab", "socket2"]
 
-io-uring = ["ntex-neon/io-uring", "dep:io-uring", "socket2"]
+polling = ["ntex-neon/polling", "dep:polling"]
+io-uring = ["ntex-neon/io-uring", "dep:io-uring"]
 
 [dependencies]
 ntex-service = "3.3"
 ntex-bytes = "0.1"
 ntex-http = "0.1"
-ntex-io = "2.11.2"
-ntex-rt = "0.4.30"
+ntex-io = "2.11"
+ntex-rt = "0.4.25"
 ntex-util = "2.5"
 
 ntex-tokio = { version = "0.5.3", optional = true }
 ntex-compio = { version = "0.2.4", optional = true }
-ntex-neon = { version = "0.1.17", optional = true }
+ntex-neon = { version = "0.1.3", optional = true }
 
 bitflags = { workspace = true }
 cfg-if = { workspace = true }
 log = { workspace = true }
 libc = { workspace = true }
 slab = { workspace = true, optional = true }
-socket2 = { workspace = true, optional = true, features = ["all"] }
+socket2 = { workspace = true, optional = true }
 thiserror = { workspace = true }
 
 # Linux specific dependencies
 [target.'cfg(target_os = "linux")'.dependencies]
 io-uring = { workspace = true, optional = true }
+polling = { workspace = true, optional = true }
 
 [dev-dependencies]
 ntex = "2"
-oneshot = "0.1"
+env_logger = "0.11"

ntex-net/src/lib.rs

@@ -14,14 +14,14 @@ cfg_if::cfg_if! {
         mod rt_impl;
         pub use self::rt_impl::{
             from_tcp_stream, from_unix_stream, tcp_connect, tcp_connect_in, unix_connect,
-            unix_connect_in, active_stream_ops
+            unix_connect_in,
         };
     } else if #[cfg(all(unix, feature = "neon"))] {
         #[path = "rt_polling/mod.rs"]
         mod rt_impl;
         pub use self::rt_impl::{
             from_tcp_stream, from_unix_stream, tcp_connect, tcp_connect_in, unix_connect,
-            unix_connect_in, active_stream_ops
+            unix_connect_in,
         };
     } else {
         pub use self::compat::*;

ntex-net/src/rt_polling/connect.rs

@@ -1,7 +1,7 @@
 use std::os::fd::{AsRawFd, RawFd};
-use std::{cell::RefCell, io, rc::Rc, task::Poll};
+use std::{cell::RefCell, collections::VecDeque, io, rc::Rc, task::Poll};
 
-use ntex_neon::driver::{DriverApi, Event, Handler};
+use ntex_neon::driver::{DriverApi, Handler, Interest};
 use ntex_neon::{syscall, Runtime};
 use ntex_util::channel::oneshot::Sender;
 use slab::Slab;
@@ -12,11 +12,13 @@ pub(crate) struct ConnectOps(Rc<ConnectOpsInner>);
 
 #[derive(Debug)]
 enum Change {
-    Event(Event),
+    Readable,
+    Writable,
     Error(io::Error),
 }
 
 struct ConnectOpsBatcher {
+    feed: VecDeque<(usize, Change)>,
     inner: Rc<ConnectOpsInner>,
 }
 
@@ -34,13 +36,16 @@ impl ConnectOps {
     pub(crate) fn current() -> Self {
         Runtime::value(|rt| {
             let mut inner = None;
-            rt.register_handler(|api| {
+            rt.driver().register(|api| {
                 let ops = Rc::new(ConnectOpsInner {
                     api,
                     connects: RefCell::new(Slab::new()),
                 });
                 inner = Some(ops.clone());
-                Box::new(ConnectOpsBatcher { inner: ops })
+                Box::new(ConnectOpsBatcher {
+                    inner: ops,
+                    feed: VecDeque::new(),
+                })
             });
 
             ConnectOps(inner.unwrap())
@@ -62,49 +67,68 @@ impl ConnectOps {
         let item = Item { fd, sender };
         let id = self.0.connects.borrow_mut().insert(item);
 
-        self.0.api.attach(fd, id as u32, Event::writable(0));
+        self.0.api.register(fd, id, Interest::Writable);
+
         Ok(id)
     }
 }
 
 impl Handler for ConnectOpsBatcher {
-    fn event(&mut self, id: usize, event: Event) {
-        log::trace!("connect-fd is readable {:?}", id);
+    fn readable(&mut self, id: usize) {
+        log::debug!("connect-fd is readable {:?}", id);
+        self.feed.push_back((id, Change::Readable));
+    }
 
-        let mut connects = self.inner.connects.borrow_mut();
-        if connects.contains(id) {
-            let item = connects.remove(id);
-            if event.writable {
-                let mut err: libc::c_int = 0;
-                let mut err_len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
-
-                let res = syscall!(libc::getsockopt(
-                    item.fd.as_raw_fd(),
-                    libc::SOL_SOCKET,
-                    libc::SO_ERROR,
-                    &mut err as *mut _ as *mut _,
-                    &mut err_len
-                ));
-
-                let res = if err == 0 {
-                    res.map(|_| ())
-                } else {
-                    Err(io::Error::from_raw_os_error(err))
-                };
-
-                self.inner.api.detach(item.fd, id as u32);
-                let _ = item.sender.send(res);
-            }
-        }
+    fn writable(&mut self, id: usize) {
+        log::debug!("connect-fd is writable {:?}", id);
+        self.feed.push_back((id, Change::Writable));
     }
 
     fn error(&mut self, id: usize, err: io::Error) {
+        self.feed.push_back((id, Change::Error(err)));
+    }
+
+    fn commit(&mut self) {
+        if self.feed.is_empty() {
+            return;
+        }
+        log::debug!("Commit connect driver changes, num: {:?}", self.feed.len());
+
         let mut connects = self.inner.connects.borrow_mut();
 
-        if connects.contains(id) {
-            let item = connects.remove(id);
-            let _ = item.sender.send(Err(err));
-            self.inner.api.detach(item.fd, id as u32);
+        for (id, change) in self.feed.drain(..) {
+            if connects.contains(id) {
+                let item = connects.remove(id);
+                match change {
+                    Change::Readable => unreachable!(),
+                    Change::Writable => {
+                        let mut err: libc::c_int = 0;
+                        let mut err_len =
+                            std::mem::size_of::<libc::c_int>() as libc::socklen_t;
+
+                        let res = syscall!(libc::getsockopt(
+                            item.fd.as_raw_fd(),
+                            libc::SOL_SOCKET,
+                            libc::SO_ERROR,
+                            &mut err as *mut _ as *mut _,
+                            &mut err_len
+                        ));
+
+                        let res = if err == 0 {
+                            res.map(|_| ())
+                        } else {
+                            Err(io::Error::from_raw_os_error(err))
+                        };
+
+                        self.inner.api.unregister_all(item.fd);
+                        let _ = item.sender.send(res);
+                    }
+                    Change::Error(err) => {
+                        let _ = item.sender.send(Err(err));
+                        self.inner.api.unregister_all(item.fd);
+                    }
+                }
+            }
         }
     }
 }

ntex-net/src/rt_polling/driver.rs

@@ -1,287 +1,344 @@
-use std::os::fd::RawFd;
-use std::{cell::Cell, cell::RefCell, future::Future, io, rc::Rc, task::Poll};
+use std::os::fd::{AsRawFd, RawFd};
+use std::{cell::Cell, collections::VecDeque, io, rc::Rc, task, task::Poll};
 
-use ntex_neon::driver::{DriverApi, Event, Handler};
+use ntex_neon::driver::{DriverApi, Handler, Interest};
 use ntex_neon::{syscall, Runtime};
 use slab::Slab;
 
 use ntex_bytes::BufMut;
 use ntex_io::IoContext;
 
-pub(crate) struct StreamCtl {
-    id: u32,
-    inner: Rc<StreamOpsInner>,
-}
-
 bitflags::bitflags! {
-    #[derive(Copy, Clone, Debug)]
+    #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
     struct Flags: u8 {
-        const RD     = 0b0000_0001;
-        const WR     = 0b0000_0010;
-        const CLOSED = 0b0001_0000;
+        const ERROR = 0b0000_0001;
+        const RD    = 0b0000_0010;
+        const WR    = 0b0000_0100;
     }
 }
 
-struct StreamItem {
+pub(crate) struct StreamCtl<T> {
+    id: usize,
+    inner: Rc<StreamOpsInner<T>>,
+}
+
+struct StreamItem<T> {
+    io: Option<T>,
     fd: RawFd,
-    flags: Flags,
-    ref_count: u16,
     context: IoContext,
+    flags: Flags,
+    ref_count: usize,
 }
 
-pub(crate) struct StreamOps(Rc<StreamOpsInner>);
+pub(crate) struct StreamOps<T>(Rc<StreamOpsInner<T>>);
 
-struct StreamOpsHandler {
-    inner: Rc<StreamOpsInner>,
+#[derive(Debug)]
+enum Change {
+    Readable,
+    Writable,
+    Error(io::Error),
 }
 
-struct StreamOpsInner {
+struct StreamOpsHandler<T> {
+    feed: VecDeque<(usize, Change)>,
+    inner: Rc<StreamOpsInner<T>>,
+}
+
+struct StreamOpsInner<T> {
     api: DriverApi,
-    delayd_drop: Cell<bool>,
-    feed: RefCell<Vec<u32>>,
-    streams: Cell<Option<Box<Slab<StreamItem>>>>,
+    feed: Cell<Option<VecDeque<usize>>>,
+    streams: Cell<Option<Box<Slab<StreamItem<T>>>>>,
 }
 
-impl StreamOps {
+impl<T: AsRawFd + 'static> StreamOps<T> {
     pub(crate) fn current() -> Self {
         Runtime::value(|rt| {
             let mut inner = None;
-            rt.register_handler(|api| {
+            rt.driver().register(|api| {
                 let ops = Rc::new(StreamOpsInner {
                     api,
-                    feed: RefCell::new(Vec::new()),
-                    delayd_drop: Cell::new(false),
+                    feed: Cell::new(Some(VecDeque::new())),
                     streams: Cell::new(Some(Box::new(Slab::new()))),
                 });
                 inner = Some(ops.clone());
-                Box::new(StreamOpsHandler { inner: ops })
+                Box::new(StreamOpsHandler {
+                    inner: ops,
+                    feed: VecDeque::new(),
+                })
             });
 
             StreamOps(inner.unwrap())
         })
     }
 
-    pub(crate) fn active_ops() -> usize {
-        Self::current().0.with(|streams| streams.len())
-    }
-
-    pub(crate) fn register(&self, fd: RawFd, context: IoContext) -> StreamCtl {
-        let stream = self.0.with(move |streams| {
-            let item = StreamItem {
-                fd,
-                context,
-                ref_count: 1,
-                flags: Flags::empty(),
-            };
+    pub(crate) fn register(&self, io: T, context: IoContext) -> StreamCtl<T> {
+        let item = StreamItem {
+            context,
+            fd: io.as_raw_fd(),
+            io: Some(io),
+            flags: Flags::empty(),
+            ref_count: 1,
+        };
+        self.with(|streams| {
+            let id = streams.insert(item);
             StreamCtl {
-                id: streams.insert(item) as u32,
+                id,
                 inner: self.0.clone(),
             }
-        });
+        })
+    }
 
-        self.0
-            .api
-            .attach(fd, stream.id, Event::new(0, false, false));
-        stream
+    fn with<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(&mut Slab<StreamItem<T>>) -> R,
+    {
+        let mut inner = self.0.streams.take().unwrap();
+        let result = f(&mut inner);
+        self.0.streams.set(Some(inner));
+        result
     }
 }
 
-impl Clone for StreamOps {
+impl<T> Clone for StreamOps<T> {
     fn clone(&self) -> Self {
         Self(self.0.clone())
     }
 }
 
-impl Handler for StreamOpsHandler {
-    fn event(&mut self, id: usize, ev: Event) {
-        self.inner.with(|streams| {
-            if !streams.contains(id) {
-                return;
-            }
-            let item = &mut streams[id];
-            if item.flags.contains(Flags::CLOSED) {
-                return;
-            }
+impl<T> Handler for StreamOpsHandler<T> {
+    fn readable(&mut self, id: usize) {
+        log::debug!("FD is readable {:?}", id);
+        self.feed.push_back((id, Change::Readable));
+    }
 
-            log::trace!("{}: Event ({:?}): {:?}", item.tag(), item.fd, ev);
-
-            let mut flags = item.flags;
-            let mut renew = Event::new(0, false, false);
-            if ev.readable {
-                if item
-                    .read(&mut flags, id as u32, &self.inner.api)
-                    .is_pending()
-                    && item.can_read()
-                {
-                    renew.readable = true;
-                    flags.insert(Flags::RD);
-                } else {
-                    flags.remove(Flags::RD);
-                }
-            } else if flags.contains(Flags::RD) {
-                renew.readable = true;
-            }
-
-            if ev.writable {
-                let result = item.context.with_write_buf(|buf| {
-                    log::trace!("{}: write {:?} s: {:?}", item.tag(), item.fd, buf.len());
-                    syscall!(break libc::write(item.fd, buf[..].as_ptr() as _, buf.len()))
-                });
-                if result.is_pending() {
-                    renew.writable = true;
-                    flags.insert(Flags::WR);
-                } else {
-                    flags.remove(Flags::WR);
-                }
-            } else if flags.contains(Flags::WR) {
-                renew.writable = true;
-            }
-            item.flags = flags;
-
-            // handle HUP
-            if ev.is_interrupt() {
-                item.close(id as u32, &self.inner.api, None, false);
-                return;
-            }
-
-            // register Event in driver
-            if !item.flags.contains(Flags::CLOSED) {
-                self.inner.api.modify(item.fd, id as u32, renew);
-            }
-
-            // delayed drops
-            if self.inner.delayd_drop.get() {
-                for id in self.inner.feed.borrow_mut().drain(..) {
-                    let item = &mut streams[id as usize];
-                    item.ref_count -= 1;
-                    if item.ref_count == 0 {
-                        let mut item = streams.remove(id as usize);
-                        log::trace!(
-                            "{}: Drop ({:?}), flags: {:?}",
-                            item.tag(),
-                            item.fd,
-                            item.flags
-                        );
-                        item.close(id, &self.inner.api, None, true);
-                    }
-                }
-                self.inner.delayd_drop.set(false);
-            }
-        });
+    fn writable(&mut self, id: usize) {
+        log::debug!("FD is writable {:?}", id);
+        self.feed.push_back((id, Change::Writable));
     }
 
     fn error(&mut self, id: usize, err: io::Error) {
-        self.inner.with(|streams| {
-            if let Some(item) = streams.get_mut(id) {
-                log::trace!(
-                    "{}: FD is failed ({:?}) err: {:?}",
-                    item.tag(),
-                    item.fd,
-                    err
-                );
-                item.close(id as u32, &self.inner.api, Some(err), false);
-            }
-        })
+        log::debug!("FD is failed {:?}, err: {:?}", id, err);
+        self.feed.push_back((id, Change::Error(err)));
     }
-}
 
-impl StreamOpsInner {
-    fn with<F, R>(&self, f: F) -> R
-    where
-        F: FnOnce(&mut Slab<StreamItem>) -> R,
-    {
-        let mut streams = self.streams.take().unwrap();
-        let result = f(&mut streams);
-        self.streams.set(Some(streams));
-        result
-    }
-}
-
-impl StreamCtl {
-    pub(crate) fn close(self) -> impl Future<Output = io::Result<()>> {
-        let id = self.id as usize;
-        let fut = self
-            .inner
-            .with(|streams| streams[id].close(self.id, &self.inner.api, None, true));
-        async move {
-            if let Some(fut) = fut {
-                fut.await
-                    .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
-                    .and_then(crate::helpers::pool_io_err)?;
-            }
-            Ok(())
+    fn commit(&mut self) {
+        if self.feed.is_empty() {
+            return;
         }
-    }
+        log::debug!("Commit changes, num: {:?}", self.feed.len());
 
-    pub(crate) fn modify(&self, rd: bool, wr: bool) -> bool {
-        self.inner.with(|streams| {
-            let item = &mut streams[self.id as usize];
-            if item.flags.contains(Flags::CLOSED) {
-                return false;
-            }
+        let mut streams = self.inner.streams.take().unwrap();
 
-            log::trace!(
-                "{}: Modify interest ({:?}) rd: {:?}, wr: {:?}",
-                item.tag(),
-                item.fd,
-                rd,
-                wr
-            );
-
-            let mut flags = item.flags;
-            let mut event = Event::new(0, false, false);
-
-            if rd {
-                if flags.contains(Flags::RD) {
-                    event.readable = true;
-                } else if item.read(&mut flags, self.id, &self.inner.api).is_pending() {
-                    event.readable = true;
-                    flags.insert(Flags::RD);
-                }
-            } else if flags.contains(Flags::RD) {
-                flags.remove(Flags::RD);
-            }
-
-            if wr {
-                if flags.contains(Flags::WR) {
-                    event.writable = true;
-                } else {
-                    let result = item.context.with_write_buf(|buf| {
-                        log::trace!(
-                            "{}: Writing ({}), buf: {:?}",
-                            item.tag(),
-                            item.fd,
-                            buf.len()
-                        );
-                        syscall!(
-                            break libc::write(item.fd, buf[..].as_ptr() as _, buf.len())
+        for (id, change) in self.feed.drain(..) {
+            match change {
+                Change::Readable => {
+                    let item = &mut streams[id];
+                    let result = item.context.with_read_buf(|buf| {
+                        let chunk = buf.chunk_mut();
+                        let b = chunk.as_mut_ptr();
+                        Poll::Ready(
+                            task::ready!(syscall!(
+                                break libc::read(item.fd, b as _, chunk.len())
+                            ))
+                            .inspect(|size| {
+                                unsafe { buf.advance_mut(*size) };
+                                log::debug!(
+                                    "{}: {:?}, SIZE: {:?}, BUF: {:?}",
+                                    item.context.tag(),
+                                    item.fd,
+                                    size,
+                                    buf
+                                );
+                            }),
                         )
                     });
 
                     if result.is_pending() {
-                        event.writable = true;
-                        flags.insert(Flags::WR);
+                        item.flags.insert(Flags::RD);
+                        self.inner.api.register(item.fd, id, Interest::Readable);
+                    } else {
+                        item.flags.remove(Flags::RD);
                     }
                 }
-            } else if flags.contains(Flags::WR) {
-                flags.remove(Flags::WR);
-            }
-            item.flags = flags;
+                Change::Writable => {
+                    let item = &mut streams[id];
+                    let result = item.context.with_write_buf(|buf| {
+                        let slice = &buf[..];
+                        syscall!(
+                            break libc::write(item.fd, slice.as_ptr() as _, slice.len())
+                        )
+                    });
 
-            if !flags.contains(Flags::CLOSED) {
-                self.inner.api.modify(item.fd, self.id, event);
-                true
-            } else {
-                false
+                    if result.is_pending() {
+                        item.flags.insert(Flags::WR);
+                        self.inner.api.register(item.fd, id, Interest::Writable);
+                    } else {
+                        item.flags.remove(Flags::WR);
+                    }
+                }
+                Change::Error(err) => {
+                    if let Some(item) = streams.get_mut(id) {
+                        item.context.stopped(Some(err));
+                        if !item.flags.contains(Flags::ERROR) {
+                            item.flags.insert(Flags::ERROR);
+                            item.flags.remove(Flags::RD | Flags::WR);
+                            self.inner.api.unregister_all(item.fd);
+                        }
+                    }
+                }
             }
-        })
+        }
+
+        // extra
+        let mut feed = self.inner.feed.take().unwrap();
+        for id in feed.drain(..) {
+            let item = &mut streams[id];
+            log::debug!("{}: Drop io ({}), {:?}", item.context.tag(), id, item.fd);
+
+            item.ref_count -= 1;
+            if item.ref_count == 0 {
+                let item = streams.remove(id);
+                if item.io.is_some() {
+                    self.inner.api.unregister_all(item.fd);
+                }
+            }
+        }
+
+        self.inner.feed.set(Some(feed));
+        self.inner.streams.set(Some(streams));
     }
 }
 
-impl Clone for StreamCtl {
+impl<T> StreamCtl<T> {
+    pub(crate) async fn close(self) -> io::Result<()> {
+        let (io, fd) =
+            self.with(|streams| (streams[self.id].io.take(), streams[self.id].fd));
+        if let Some(io) = io {
+            std::mem::forget(io);
+
+            ntex_rt::spawn_blocking(move || syscall!(libc::close(fd)))
+                .await
+                .map_err(|e| io::Error::new(io::ErrorKind::Other, e))
+                .and_then(crate::helpers::pool_io_err)?;
+        }
+        Ok(())
+    }
+
+    pub(crate) fn with_io<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(Option<&T>) -> R,
+    {
+        self.with(|streams| f(streams[self.id].io.as_ref()))
+    }
+
+    pub(crate) fn pause_all(&self) {
+        self.with(|streams| {
+            let item = &mut streams[self.id];
+
+            if item.flags.intersects(Flags::RD | Flags::WR) {
+                log::debug!(
+                    "{}: Pause all io ({}), {:?}",
+                    item.context.tag(),
+                    self.id,
+                    item.fd
+                );
+                item.flags.remove(Flags::RD | Flags::WR);
+                self.inner.api.unregister_all(item.fd);
+            }
+        })
+    }
+
+    pub(crate) fn pause_read(&self) {
+        self.with(|streams| {
+            let item = &mut streams[self.id];
+
+            log::debug!(
+                "{}: Pause io read ({}), {:?}",
+                item.context.tag(),
+                self.id,
+                item.fd
+            );
+            if item.flags.contains(Flags::RD) {
+                item.flags.remove(Flags::RD);
+                self.inner.api.unregister(item.fd, Interest::Readable);
+            }
+        })
+    }
+
+    pub(crate) fn resume_read(&self) {
+        self.with(|streams| {
+            let item = &mut streams[self.id];
+
+            log::debug!(
+                "{}: Resume io read ({}), {:?}",
+                item.context.tag(),
+                self.id,
+                item.fd
+            );
+            if !item.flags.contains(Flags::RD) {
+                item.flags.insert(Flags::RD);
+                self.inner
+                    .api
+                    .register(item.fd, self.id, Interest::Readable);
+            }
+        })
+    }
+
+    pub(crate) fn resume_write(&self) {
+        self.with(|streams| {
+            let item = &mut streams[self.id];
+
+            if !item.flags.contains(Flags::WR) {
+                log::debug!(
+                    "{}: Resume io write ({}), {:?}",
+                    item.context.tag(),
+                    self.id,
+                    item.fd
+                );
+                let result = item.context.with_write_buf(|buf| {
+                    log::debug!(
+                        "{}: Writing io ({}), buf: {:?}",
+                        item.context.tag(),
+                        self.id,
+                        buf.len()
+                    );
+
+                    let slice = &buf[..];
+                    syscall!(break libc::write(item.fd, slice.as_ptr() as _, slice.len()))
+                });
+
+                if result.is_pending() {
+                    log::debug!(
+                        "{}: Write is pending ({}), {:?}",
+                        item.context.tag(),
+                        self.id,
+                        item.context.flags()
+                    );
+
+                    item.flags.insert(Flags::WR);
+                    self.inner
+                        .api
+                        .register(item.fd, self.id, Interest::Writable);
+                }
+            }
+        })
+    }
+
+    fn with<F, R>(&self, f: F) -> R
+    where
+        F: FnOnce(&mut Slab<StreamItem<T>>) -> R,
+    {
+        let mut inner = self.inner.streams.take().unwrap();
+        let result = f(&mut inner);
+        self.inner.streams.set(Some(inner));
+        result
+    }
+}
+
+impl<T> Clone for StreamCtl<T> {
     fn clone(&self) -> Self {
-        self.inner.with(|streams| {
-            streams[self.id as usize].ref_count += 1;
+        self.with(|streams| {
+            streams[self.id].ref_count += 1;
             Self {
                 id: self.id,
                 inner: self.inner.clone(),
@@ -290,125 +347,28 @@ impl Clone for StreamCtl {
     }
 }
 
-impl Drop for StreamCtl {
+impl<T> Drop for StreamCtl<T> {
     fn drop(&mut self) {
         if let Some(mut streams) = self.inner.streams.take() {
-            let id = self.id as usize;
-            streams[id].ref_count -= 1;
-            if streams[id].ref_count == 0 {
-                let mut item = streams.remove(id);
-                log::trace!(
-                    "{}: Drop io ({:?}), flags: {:?}",
-                    item.tag(),
-                    item.fd,
-                    item.flags
-                );
-                item.close(self.id, &self.inner.api, None, true);
+            log::debug!(
+                "{}: Drop io ({}), {:?}",
+                streams[self.id].context.tag(),
+                self.id,
+                streams[self.id].fd
+            );
+
+            streams[self.id].ref_count -= 1;
+            if streams[self.id].ref_count == 0 {
+                let item = streams.remove(self.id);
+                if item.io.is_some() {
+                    self.inner.api.unregister_all(item.fd);
+                }
             }
             self.inner.streams.set(Some(streams));
         } else {
-            self.inner.delayd_drop.set(true);
-            self.inner.feed.borrow_mut().push(self.id);
-        }
-    }
-}
-
-impl StreamItem {
-    fn tag(&self) -> &'static str {
-        self.context.tag()
-    }
-
-    fn can_read(&self) -> bool {
-        self.context.is_read_ready()
-    }
-
-    fn read(&mut self, flags: &mut Flags, id: u32, api: &DriverApi) -> Poll<()> {
-        let mut close = false;
-
-        let result = self.context.with_read_buf(|buf, hw, lw| {
-            let mut total = 0;
-            loop {
-                // make sure we've got room
-                if buf.remaining_mut() < lw {
-                    buf.reserve(hw);
-                }
-
-                let chunk = buf.chunk_mut();
-                let chunk_len = chunk.len();
-                let chunk_ptr = chunk.as_mut_ptr();
-
-                let res = syscall!(break libc::read(self.fd, chunk_ptr as _, chunk_len));
-                if let Poll::Ready(Ok(size)) = res {
-                    unsafe { buf.advance_mut(size) };
-                    total += size;
-                    if size == chunk_len {
-                        continue;
-                    }
-                }
-
-                log::trace!(
-                    "{}: Read fd ({:?}), size: {:?}, cap: {:?}, result: {:?}",
-                    self.tag(),
-                    self.fd,
-                    total,
-                    buf.remaining_mut(),
-                    res
-                );
-
-                return match res {
-                    Poll::Ready(Err(err)) => {
-                        close = true;
-                        (total, Poll::Ready(Err(err)))
-                    }
-                    Poll::Ready(Ok(size)) => {
-                        if size == 0 {
-                            close = true;
-                        }
-                        (total, Poll::Ready(Ok(())))
-                    }
-                    Poll::Pending => (total, Poll::Pending),
-                };
-            }
-        });
-
-        if close {
-            flags.insert(Flags::CLOSED);
-            self.close(id, api, None, false);
-        }
-        result
-    }
-
-    fn close(
-        &mut self,
-        id: u32,
-        api: &DriverApi,
-        error: Option<io::Error>,
-        shutdown: bool,
-    ) -> Option<ntex_rt::JoinHandle<io::Result<i32>>> {
-        if !self.flags.contains(Flags::CLOSED) {
-            log::trace!(
-                "{}: Closing ({}) sh: {:?}, flags: {:?}, ctx: {:?}",
-                self.tag(),
-                self.fd,
-                shutdown,
-                self.flags,
-                self.context.flags()
-            );
-            self.flags.insert(Flags::CLOSED);
-            if !self.context.is_stopped() {
-                self.context.stopped(error);
-            }
-
-            let fd = self.fd;
-            api.detach(fd, id);
-            Some(ntex_rt::spawn_blocking(move || {
-                if shutdown {
-                    let _ = syscall!(libc::shutdown(fd, libc::SHUT_RDWR));
-                }
-                syscall!(libc::close(fd))
-            }))
-        } else {
-            None
+            let mut feed = self.inner.feed.take().unwrap();
+            feed.push_back(self.id);
+            self.inner.feed.set(Some(feed));
         }
     }
 }

ntex-net/src/rt_polling/io.rs

@@ -1,4 +1,4 @@
-use std::{any, future::poll_fn, mem, os::fd::AsRawFd, task::Poll};
+use std::{any, future::poll_fn, task::Poll};
 
 use ntex_io::{
     types, Handle, IoContext, IoStream, ReadContext, ReadStatus, WriteContext, WriteStatus,
@@ -12,10 +12,11 @@ impl IoStream for super::TcpStream {
     fn start(self, read: ReadContext, _: WriteContext) -> Option<Box<dyn Handle>> {
         let io = self.0;
         let context = read.context();
-        let ctl = StreamOps::current().register(io.as_raw_fd(), context.clone());
+        let ctl = StreamOps::current().register(io, context.clone());
+        let ctl2 = ctl.clone();
         spawn(async move { run(ctl, context).await });
 
-        Some(Box::new(HandleWrapper(Some(io))))
+        Some(Box::new(HandleWrapper(ctl2)))
     }
 }
 
@@ -23,20 +24,19 @@ impl IoStream for super::UnixStream {
     fn start(self, read: ReadContext, _: WriteContext) -> Option<Box<dyn Handle>> {
         let io = self.0;
         let context = read.context();
-        let ctl = StreamOps::current().register(io.as_raw_fd(), context.clone());
+        let ctl = StreamOps::current().register(io, context.clone());
         spawn(async move { run(ctl, context).await });
-        mem::forget(io);
 
         None
     }
 }
 
-struct HandleWrapper(Option<Socket>);
+struct HandleWrapper(StreamCtl<Socket>);
 
 impl Handle for HandleWrapper {
     fn query(&self, id: any::TypeId) -> Option<Box<dyn any::Any>> {
         if id == any::TypeId::of::<types::PeerAddr>() {
-            let addr = self.0.as_ref().unwrap().peer_addr().ok();
+            let addr = self.0.with_io(|io| io.and_then(|io| io.peer_addr().ok()));
             if let Some(addr) = addr.and_then(|addr| addr.as_socket()) {
                 return Some(Box::new(types::PeerAddr(addr)));
             }
@@ -45,41 +45,30 @@ impl Handle for HandleWrapper {
     }
 }
 
-impl Drop for HandleWrapper {
-    fn drop(&mut self) {
-        mem::forget(self.0.take());
-    }
-}
-
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 enum Status {
     Shutdown,
     Terminate,
 }
 
-async fn run(ctl: StreamCtl, context: IoContext) {
+async fn run<T>(ctl: StreamCtl<T>, context: IoContext) {
     // Handle io read readiness
     let st = poll_fn(|cx| {
-        let mut modify = false;
-        let mut readable = false;
-        let mut writable = false;
         let read = match context.poll_read_ready(cx) {
             Poll::Ready(ReadStatus::Ready) => {
-                modify = true;
-                readable = true;
+                ctl.resume_read();
                 Poll::Pending
             }
             Poll::Ready(ReadStatus::Terminate) => Poll::Ready(()),
             Poll::Pending => {
-                modify = true;
+                ctl.pause_read();
                 Poll::Pending
             }
         };
 
         let write = match context.poll_write_ready(cx) {
             Poll::Ready(WriteStatus::Ready) => {
-                modify = true;
-                writable = true;
+                ctl.resume_write();
                 Poll::Pending
             }
             Poll::Ready(WriteStatus::Shutdown) => Poll::Ready(Status::Shutdown),
@@ -87,10 +76,6 @@ async fn run(ctl: StreamCtl, context: IoContext) {
             Poll::Pending => Poll::Pending,
         };
 
-        if modify && !ctl.modify(readable, writable) {
-            return Poll::Ready(Status::Terminate);
-        }
-
         if read.is_pending() && write.is_pending() {
             Poll::Pending
         } else if write.is_ready() {
@@ -101,10 +86,11 @@ async fn run(ctl: StreamCtl, context: IoContext) {
     })
     .await;
 
-    if st != Status::Terminate && ctl.modify(false, true) {
-        context.shutdown(st == Status::Shutdown).await;
-    }
-    if !context.is_stopped() {
-        context.stopped(ctl.close().await.err());
-    }
+    ctl.resume_write();
+    context.shutdown(st == Status::Shutdown).await;
+
+    ctl.pause_all();
+    let result = ctl.close().await;
+
+    context.stopped(result.err());
 }

ntex-net/src/rt_polling/mod.rs

@@ -8,9 +8,6 @@ pub(crate) mod connect;
 mod driver;
 mod io;
 
-#[cfg(not(target_pointer_width = "64"))]
-compile_error!("Only 64bit platforms are supported");
-
 /// Tcp stream wrapper for neon TcpStream
 struct TcpStream(socket2::Socket);
 
@@ -67,68 +64,3 @@ pub fn from_unix_stream(stream: std::os::unix::net::UnixStream) -> Result<Io> {
         Socket::from(stream),
     )?)))
 }
-
-#[doc(hidden)]
-/// Get number of active Io objects
-pub fn active_stream_ops() -> usize {
-    self::driver::StreamOps::active_ops()
-}
-
-#[cfg(all(target_os = "linux", feature = "neon"))]
-#[cfg(test)]
-mod tests {
-    use ntex::{io::Io, time::sleep, time::Millis, util::PoolId};
-    use std::sync::{Arc, Mutex};
-
-    use crate::connect::Connect;
-
-    const DATA: &[u8] = b"Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World \
-                         Hello World Hello World Hello World Hello World Hello World";
-
-    // #[ntex::test]
-    async fn idle_disconnect() {
-        PoolId::P5.set_read_params(24, 12);
-        let (tx, rx) = ::oneshot::channel();
-        let tx = Arc::new(Mutex::new(Some(tx)));
-
-        let server = ntex::server::test_server(move || {
-            let tx = tx.clone();
-            ntex_service::fn_service(move |io: Io<_>| {
-                tx.lock().unwrap().take().unwrap().send(()).unwrap();
-
-                async move {
-                    io.write(DATA).unwrap();
-                    sleep(Millis(250)).await;
-                    io.close();
-                    Ok::<_, ()>(())
-                }
-            })
-        });
-
-        let msg = Connect::new(server.addr());
-        let io = crate::connect::connect(msg).await.unwrap();
-        io.set_memory_pool(PoolId::P5.into());
-        rx.await.unwrap();
-
-        io.on_disconnect().await;
-    }
-}

ntex-net/src/rt_uring/connect.rs

@@ -31,11 +31,7 @@ impl ConnectOps {
     pub(crate) fn current() -> Self {
         Runtime::value(|rt| {
             let mut inner = None;
-            rt.register_handler(|api| {
-                if !api.is_supported(opcode::Connect::CODE) {
-                    panic!("opcode::Connect is required for io-uring support");
-                }
-
+            rt.driver().register(|api| {
                 let ops = Rc::new(ConnectOpsInner {
                     api,
                     ops: RefCell::new(Slab::new()),
@@ -72,14 +68,14 @@ impl ConnectOps {
 
 impl Handler for ConnectOpsHandler {
     fn canceled(&mut self, user_data: usize) {
-        log::trace!("connect-op is canceled {:?}", user_data);
+        log::debug!("connect-op is canceled {:?}", user_data);
 
         self.inner.ops.borrow_mut().remove(user_data);
     }
 
     fn completed(&mut self, user_data: usize, flags: u32, result: io::Result<i32>) {
         let (addr, tx) = self.inner.ops.borrow_mut().remove(user_data);
-        log::trace!(
+        log::debug!(
             "connect-op is completed {:?} result: {:?}, addr: {:?}",
             user_data,
             result,

ntex-net/src/rt_uring/driver.rs

@@ -13,32 +13,15 @@ pub(crate) struct StreamCtl<T> {
     inner: Rc<StreamOpsInner<T>>,
 }
 
-bitflags::bitflags! {
-    #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
-    struct Flags: u8 {
-        const RD_CANCELING = 0b0000_0001;
-        const RD_REISSUE   = 0b0000_0010;
-        const WR_CANCELING = 0b0001_0000;
-        const WR_REISSUE   = 0b0010_0000;
-    }
-}
-
 struct StreamItem<T> {
     io: Option<T>,
     fd: Fd,
     context: IoContext,
     ref_count: usize,
-    flags: Flags,
     rd_op: Option<NonZeroU32>,
     wr_op: Option<NonZeroU32>,
 }
 
-impl<T> StreamItem<T> {
-    fn tag(&self) -> &'static str {
-        self.context.tag()
-    }
-}
-
 enum Operation {
     Recv {
         id: usize,
@@ -77,17 +60,7 @@ impl<T: os::fd::AsRawFd + 'static> StreamOps<T> {
     pub(crate) fn current() -> Self {
         Runtime::value(|rt| {
             let mut inner = None;
-            rt.register_handler(|api| {
-                if !api.is_supported(opcode::Recv::CODE) {
-                    panic!("opcode::Recv is required for io-uring support");
-                }
-                if !api.is_supported(opcode::Send::CODE) {
-                    panic!("opcode::Send is required for io-uring support");
-                }
-                if !api.is_supported(opcode::Close::CODE) {
-                    panic!("opcode::Close is required for io-uring support");
-                }
-
+            rt.driver().register(|api| {
                 let mut ops = Slab::new();
                 ops.insert(Operation::Nop);
 
@@ -115,7 +88,6 @@ impl<T: os::fd::AsRawFd + 'static> StreamOps<T> {
             ref_count: 1,
             rd_op: None,
             wr_op: None,
-            flags: Flags::empty(),
         };
         let id = self.0.storage.borrow_mut().streams.insert(item);
         StreamCtl {
@@ -124,10 +96,6 @@ impl<T: os::fd::AsRawFd + 'static> StreamOps<T> {
         }
     }
 
-    pub(crate) fn active_ops() -> usize {
-        Self::current().with(|st| st.streams.len())
-    }
-
     fn with<F, R>(&self, f: F) -> R
     where
         F: FnOnce(&mut StreamOpsStorage<T>) -> R,
@@ -148,35 +116,17 @@ impl<T> Handler for StreamOpsHandler<T> {
 
         match storage.ops.remove(user_data) {
             Operation::Recv { id, buf, context } => {
-                log::trace!("{}: Recv canceled {:?}", context.tag(), id);
+                log::debug!("{}: Recv canceled {:?}", context.tag(), id,);
                 context.release_read_buf(buf);
                 if let Some(item) = storage.streams.get_mut(id) {
                     item.rd_op.take();
-                    item.flags.remove(Flags::RD_CANCELING);
-                    if item.flags.contains(Flags::RD_REISSUE) {
-                        item.flags.remove(Flags::RD_REISSUE);
-
-                        let result = storage.recv(id, Some(context));
-                        if let Some((id, op)) = result {
-                            self.inner.api.submit(id, op);
-                        }
-                    }
                 }
             }
             Operation::Send { id, buf, context } => {
-                log::trace!("{}: Send canceled: {:?}", context.tag(), id);
+                log::debug!("{}: Send canceled: {:?}", context.tag(), id);
                 context.release_write_buf(buf);
                 if let Some(item) = storage.streams.get_mut(id) {
                     item.wr_op.take();
-                    item.flags.remove(Flags::WR_CANCELING);
-                    if item.flags.contains(Flags::WR_REISSUE) {
-                        item.flags.remove(Flags::WR_REISSUE);
-
-                        let result = storage.send(id, Some(context));
-                        if let Some((id, op)) = result {
-                            self.inner.api.submit(id, op);
-                        }
-                    }
                 }
             }
             Operation::Nop | Operation::Close { .. } => {}
@@ -200,12 +150,13 @@ impl<T> Handler for StreamOpsHandler<T> {
 
                 // reset op reference
                 if let Some(item) = storage.streams.get_mut(id) {
-                    log::trace!(
-                        "{}: Recv completed {:?}, res: {:?}, buf({})",
+                    log::debug!(
+                        "{}: Recv completed {:?}, res: {:?}, buf({}): {:?}",
                         context.tag(),
                         item.fd,
                         result,
-                        buf.remaining_mut()
+                        buf.remaining_mut(),
+                        buf,
                     );
                     item.rd_op.take();
                 }
@@ -217,29 +168,26 @@ impl<T> Handler for StreamOpsHandler<T> {
                         self.inner.api.submit(id, op);
                     }
                 } else {
-                    log::trace!("{}: Recv to pause", tag);
+                    log::debug!("{}: Recv to pause", tag);
                 }
             }
             Operation::Send { id, buf, context } => {
                 // reset op reference
-                let fd = if let Some(item) = storage.streams.get_mut(id) {
-                    log::trace!(
-                        "{}: Send completed: {:?}, res: {:?}, buf({})",
+                if let Some(item) = storage.streams.get_mut(id) {
+                    log::debug!(
+                        "{}: Send completed: {:?}, res: {:?}",
                         context.tag(),
                         item.fd,
-                        result,
-                        buf.len()
+                        result
                     );
                     item.wr_op.take();
-                    Some(item.fd)
-                } else {
-                    None
-                };
+                }
 
                 // set read buf
-                let result = context.set_write_buf(result.map(|size| size as usize), buf);
-                if result.is_pending() {
-                    log::trace!("{}: Need to send more: {:?}", context.tag(), fd);
+                if context
+                    .set_write_buf(result.map(|size| size as usize), buf)
+                    .is_pending()
+                {
                     if let Some((id, op)) = storage.send(id, Some(context)) {
                         self.inner.api.submit(id, op);
                     }
@@ -259,7 +207,7 @@ impl<T> Handler for StreamOpsHandler<T> {
             if storage.streams[id].ref_count == 0 {
                 let mut item = storage.streams.remove(id);
 
-                log::trace!("{}: Drop io ({}), {:?}", item.tag(), id, item.fd);
+                log::debug!("{}: Drop io ({}), {:?}", item.context.tag(), id, item.fd);
 
                 if let Some(io) = item.io.take() {
                     mem::forget(io);
@@ -281,11 +229,12 @@ impl<T> StreamOpsStorage<T> {
 
         if item.rd_op.is_none() {
             if let Poll::Ready(mut buf) = item.context.get_read_buf() {
-                log::trace!(
-                    "{}: Recv resume ({}), {:?} rem: {:?}",
-                    item.tag(),
+                log::debug!(
+                    "{}: Recv resume ({}), {:?} - {:?} = {:?}",
+                    item.context.tag(),
                     id,
                     item.fd,
+                    buf,
                     buf.remaining_mut()
                 );
 
@@ -303,8 +252,6 @@ impl<T> StreamOpsStorage<T> {
                 item.rd_op = NonZeroU32::new(op_id as u32);
                 return Some((op_id as u32, op));
             }
-        } else if item.flags.contains(Flags::RD_CANCELING) {
-            item.flags.insert(Flags::RD_REISSUE);
         }
         None
     }
@@ -314,12 +261,12 @@ impl<T> StreamOpsStorage<T> {
 
         if item.wr_op.is_none() {
             if let Poll::Ready(buf) = item.context.get_write_buf() {
-                log::trace!(
-                    "{}: Send resume ({}), {:?} len: {:?}",
-                    item.tag(),
+                log::debug!(
+                    "{}: Send resume ({}), {:?} {:?}",
+                    item.context.tag(),
                     id,
                     item.fd,
-                    buf.len()
+                    buf
                 );
 
                 let slice = buf.chunk();
@@ -336,8 +283,6 @@ impl<T> StreamOpsStorage<T> {
                 item.wr_op = NonZeroU32::new(op_id as u32);
                 return Some((op_id as u32, op));
             }
-        } else if item.flags.contains(Flags::WR_CANCELING) {
-            item.flags.insert(Flags::WR_REISSUE);
         }
         None
     }
@@ -405,11 +350,13 @@ impl<T> StreamCtl<T> {
         let item = &mut storage.streams[self.id];
 
         if let Some(rd_op) = item.rd_op {
-            if !item.flags.contains(Flags::RD_CANCELING) {
-                log::trace!("{}: Recv to pause ({}), {:?}", item.tag(), self.id, item.fd);
-                item.flags.insert(Flags::RD_CANCELING);
-                self.inner.api.cancel(rd_op.get());
-            }
+            log::debug!(
+                "{}: Recv to pause ({}), {:?}",
+                item.context.tag(),
+                self.id,
+                item.fd
+            );
+            self.inner.api.cancel(rd_op.get());
         }
     }
 }
@@ -431,7 +378,12 @@ impl<T> Drop for StreamCtl<T> {
             if storage.streams[self.id].ref_count == 0 {
                 let mut item = storage.streams.remove(self.id);
                 if let Some(io) = item.io.take() {
-                    log::trace!("{}: Close io ({}), {:?}", item.tag(), self.id, item.fd);
+                    log::debug!(
+                        "{}: Close io ({}), {:?}",
+                        item.context.tag(),
+                        self.id,
+                        item.fd
+                    );
                     mem::forget(io);
 
                     let id = storage.ops.insert(Operation::Close { tx: None });

ntex-net/src/rt_uring/mod.rs

@@ -64,9 +64,3 @@ pub fn from_unix_stream(stream: std::os::unix::net::UnixStream) -> Result<Io> {
         Socket::from(stream),
     )?)))
 }
-
-#[doc(hidden)]
-/// Get number of active Io objects
-pub fn active_stream_ops() -> usize {
-    self::driver::StreamOps::<socket2::Socket>::active_ops()
-}

ntex-rt/CHANGES.md

@@ -1,9 +1,5 @@
 # Changes
 
-## [0.4.29] - 2025-03-26
-
-* Add Arbiter::get_value() helper method
-
 ## [0.4.27] - 2025-03-14
 
 * Add srbiters pings ttl

ntex-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-rt"
-version = "0.4.30"
+version = "0.4.28"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -32,8 +32,8 @@ neon = ["ntex-neon"]
 [dependencies]
 async-channel = "2"
 futures-timer = "3.0"
-oneshot = "0.1"
 log = "0.4"
+oneshot = "0.1"
 
 compio-driver = { version = "0.6", optional = true }
 compio-runtime = { version = "0.6", optional = true }
@@ -42,4 +42,7 @@ tok-io = { version = "1", package = "tokio", default-features = false, features
     "net",
 ], optional = true }
 
-ntex-neon = { version = "0.1.17", optional = true }
+ntex-neon = { version = "0.1.1", optional = true }
+
+[dev-dependencies]
+env_logger = "0.11"

ntex-rt/src/arbiter.rs

@@ -286,25 +286,6 @@ impl Arbiter {
         })
     }
 
-    /// Get a type previously inserted to this runtime or create new one.
-    pub fn get_value<T, F>(f: F) -> T
-    where
-        T: Clone + 'static,
-        F: FnOnce() -> T,
-    {
-        STORAGE.with(move |cell| {
-            let mut st = cell.borrow_mut();
-            if let Some(boxed) = st.get(&TypeId::of::<T>()) {
-                if let Some(val) = (&**boxed as &(dyn Any + 'static)).downcast_ref::<T>() {
-                    return val.clone();
-                }
-            }
-            let val = f();
-            st.insert(TypeId::of::<T>(), Box::new(val.clone()));
-            val
-        })
-    }
-
     /// Wait for the event loop to stop by joining the underlying thread (if have Some).
     pub fn join(&mut self) -> thread::Result<()> {
         if let Some(thread_handle) = self.thread_handle.take() {
@@ -374,7 +355,6 @@ mod tests {
         assert!(Arbiter::get_item::<&'static str, _, _>(|s| *s == "test"));
         assert!(Arbiter::get_mut_item::<&'static str, _, _>(|s| *s == "test"));
         assert!(Arbiter::contains_item::<&'static str>());
-        assert!(Arbiter::get_value(|| 64u64) == 64);
         assert!(format!("{:?}", Arbiter::current()).contains("Arbiter"));
     }
 }

ntex-rt/src/lib.rs

@@ -112,8 +112,6 @@ mod tokio {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
-    #[doc(hidden)]
-    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> tok_io::task::JoinHandle<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -198,8 +196,6 @@ mod compio {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
-    #[doc(hidden)]
-    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> JoinHandle<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -265,7 +261,7 @@ mod neon {
         let rt = Runtime::new().unwrap();
         log::info!(
             "Starting neon runtime, driver {:?}",
-            rt.driver_type().name()
+            rt.driver().tp().name()
         );
 
         rt.block_on(fut);
@@ -327,8 +323,6 @@ mod neon {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
-    #[doc(hidden)]
-    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> Task<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -383,7 +377,7 @@ mod neon {
 
     impl<T> JoinHandle<T> {
         pub fn is_finished(&self) -> bool {
-            self.fut.is_none()
+            false
         }
     }
 

ntex-server/CHANGES.md

@@ -1,13 +1,5 @@
 # Changes
 
-## [2.7.3] - 2025-03-28
-
-* Better worker availability handling
-
-## [2.7.2] - 2025-03-27
-
-* Handle paused state
-
 ## [2.7.1] - 2025-02-28
 
 * Fix set core affinity out of worker start #508

ntex-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-server"
-version = "2.7.4"
+version = "2.7.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Server for ntex framework"
 keywords = ["network", "framework", "async", "futures"]
@@ -22,13 +22,13 @@ ntex-service = "3.4"
 ntex-rt = "0.4"
 ntex-util = "2.8"
 
-async-channel = { workspace = true }
-atomic-waker = { workspace = true }
-core_affinity = { workspace = true }
-oneshot = { workspace = true }
-polling = { workspace = true }
-log = { workspace = true }
-socket2 = { workspace = true }
+async-channel = "2"
+async-broadcast = "0.7"
+core_affinity = "0.8"
+polling = "3.3"
+log = "0.4"
+socket2 = "0.5"
+oneshot = { version = "0.1", default-features = false, features = ["async"] }
 
 [dev-dependencies]
 ntex = "2"

ntex-server/src/manager.rs

@@ -139,6 +139,7 @@ impl<F: ServerConfiguration> ServerManager<F> {
 fn start_worker<F: ServerConfiguration>(mgr: ServerManager<F>, cid: Option<CoreId>) {
     let _ = ntex_rt::spawn(async move {
         let id = mgr.next_id();
+
         let mut wrk = Worker::start(id, mgr.factory(), cid);
 
         loop {
@@ -180,7 +181,7 @@ impl<F: ServerConfiguration> HandleCmdState<F> {
     fn process(&mut self, mut item: F::Item) {
         loop {
             if !self.workers.is_empty() {
-                if self.next >= self.workers.len() {
+                if self.next > self.workers.len() {
                     self.next = self.workers.len() - 1;
                 }
                 match self.workers[self.next].send(item) {
@@ -211,9 +212,10 @@ impl<F: ServerConfiguration> HandleCmdState<F> {
         match upd {
             Update::Available(worker) => {
                 self.workers.push(worker);
-                self.workers.sort();
                 if self.workers.len() == 1 {
                     self.mgr.resume();
+                } else {
+                    self.workers.sort();
                 }
             }
             Update::Unavailable(worker) => {
@@ -232,9 +234,6 @@ impl<F: ServerConfiguration> HandleCmdState<F> {
                 if let Err(item) = self.workers[0].send(item) {
                     self.backlog.push_back(item);
                     self.workers.remove(0);
-                    if self.workers.is_empty() {
-                        self.mgr.pause();
-                    }
                     break;
                 }
             }

ntex-server/src/net/accept.rs

@@ -92,14 +92,12 @@ impl AcceptLoop {
 
     /// Start accept loop
     pub fn start(mut self, socks: Vec<(Token, Listener)>, srv: Server) {
-        let (tx, rx_start) = oneshot::channel();
         let (rx, poll) = self
             .inner
             .take()
             .expect("AcceptLoop cannot be used multiple times");
 
         Accept::start(
-            tx,
             rx,
             poll,
             socks,
@@ -107,8 +105,6 @@ impl AcceptLoop {
             self.notify.clone(),
             self.status_handler.take(),
         );
-
-        let _ = rx_start.recv();
     }
 }
 
@@ -125,7 +121,6 @@ impl fmt::Debug for AcceptLoop {
 struct Accept {
     poller: Arc<Poller>,
     rx: mpsc::Receiver<AcceptorCommand>,
-    tx: Option<oneshot::Sender<()>>,
     sockets: Vec<ServerSocketInfo>,
     srv: Server,
     notify: AcceptNotify,
@@ -136,7 +131,6 @@ struct Accept {
 
 impl Accept {
     fn start(
-        tx: oneshot::Sender<()>,
         rx: mpsc::Receiver<AcceptorCommand>,
         poller: Arc<Poller>,
         socks: Vec<(Token, Listener)>,
@@ -151,12 +145,11 @@ impl Accept {
             .name("ntex-server accept loop".to_owned())
             .spawn(move || {
                 System::set_current(sys);
-                Accept::new(tx, rx, poller, socks, srv, notify, status_handler).poll()
+                Accept::new(rx, poller, socks, srv, notify, status_handler).poll()
             });
     }
 
     fn new(
-        tx: oneshot::Sender<()>,
         rx: mpsc::Receiver<AcceptorCommand>,
         poller: Arc<Poller>,
         socks: Vec<(Token, Listener)>,
@@ -182,7 +175,6 @@ impl Accept {
             notify,
             srv,
             status_handler,
-            tx: Some(tx),
             backpressure: true,
             backlog: VecDeque::new(),
         }
@@ -200,23 +192,19 @@ impl Accept {
         // Create storage for events
         let mut events = Events::with_capacity(NonZeroUsize::new(512).unwrap());
 
-        let mut timeout = Some(Duration::ZERO);
         loop {
-            if let Err(e) = self.poller.wait(&mut events, timeout) {
-                if e.kind() != io::ErrorKind::Interrupted {
+            if let Err(e) = self.poller.wait(&mut events, None) {
+                if e.kind() == io::ErrorKind::Interrupted {
+                    continue;
+                } else {
                     panic!("Cannot wait for events in poller: {}", e)
                 }
-            } else if timeout.is_some() {
-                timeout = None;
-                let _ = self.tx.take().unwrap().send(());
             }
 
-            for idx in 0..self.sockets.len() {
-                if self.sockets[idx].registered.get() {
-                    let readd = self.accept(idx);
-                    if readd {
-                        self.add_source(idx);
-                    }
+            for event in events.iter() {
+                let readd = self.accept(event.key);
+                if readd {
+                    self.add_source(event.key);
                 }
             }
 

ntex-server/src/net/service.rs

@@ -114,7 +114,7 @@ impl ServiceFactory<Connection> for StreamService {
         for info in &self.services {
             match info.factory.create(()).await {
                 Ok(svc) => {
-                    log::trace!("Constructed server service for {:?}", info.tokens);
+                    log::debug!("Constructed server service for {:?}", info.tokens);
                     services.push(svc);
                     let idx = services.len() - 1;
                     for (token, tag) in &info.tokens {

ntex-server/src/net/test.rs

@@ -59,19 +59,17 @@ where
                 .workers(1)
                 .disable_signals()
                 .run();
-
-            ntex_rt::spawn(async move {
-                ntex_util::time::sleep(ntex_util::time::Millis(75)).await;
-                tx.send((system, local_addr, server))
-                    .expect("Failed to send Server to TestServer");
-            });
-
+            tx.send((system, local_addr, server))
+                .expect("Failed to send Server to TestServer");
             Ok(())
         })
     });
 
     let (system, addr, server) = rx.recv().unwrap();
 
+    // wait for server
+    thread::sleep(std::time::Duration::from_millis(50));
+
     TestServer {
         addr,
         server,

ntex-server/src/wrk.rs

@@ -2,8 +2,8 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::task::{ready, Context, Poll};
 use std::{cmp, future::poll_fn, future::Future, hash, pin::Pin, sync::Arc};
 
+use async_broadcast::{self as bus, broadcast};
 use async_channel::{unbounded, Receiver, Sender};
-use atomic_waker::AtomicWaker;
 use core_affinity::CoreId;
 
 use ntex_rt::{spawn, Arbiter};
@@ -99,10 +99,10 @@ impl<T> Worker<T> {
 
                 log::debug!("Creating server instance in {:?}", id);
                 let factory = cfg.create().await;
+                log::debug!("Server instance has been created in {:?}", id);
 
                 match create(id, rx1, rx2, factory, avail_tx).await {
                     Ok((svc, wrk)) => {
-                        log::debug!("Server instance has been created in {:?}", id);
                         run_worker(svc, wrk).await;
                     }
                     Err(e) => {
@@ -151,8 +151,10 @@ impl<T> Worker<T> {
         if self.failed.load(Ordering::Acquire) {
             WorkerStatus::Failed
         } else {
-            self.avail.wait_for_update().await;
-            if self.avail.failed() {
+            // cleanup updates
+            while self.avail.notify.try_recv().is_ok() {}
+
+            if self.avail.notify.recv_direct().await.is_err() {
                 self.failed.store(true, Ordering::Release);
             }
             self.status()
@@ -194,85 +196,52 @@ impl Future for WorkerStop {
 
 #[derive(Debug, Clone)]
 struct WorkerAvailability {
-    inner: Arc<Inner>,
+    notify: bus::Receiver<()>,
+    available: Arc<AtomicBool>,
 }
 
 #[derive(Debug, Clone)]
 struct WorkerAvailabilityTx {
-    inner: Arc<Inner>,
-}
-
-#[derive(Debug)]
-struct Inner {
-    waker: AtomicWaker,
-    updated: AtomicBool,
-    available: AtomicBool,
-    failed: AtomicBool,
+    notify: bus::Sender<()>,
+    available: Arc<AtomicBool>,
 }
 
 impl WorkerAvailability {
     fn create() -> (Self, WorkerAvailabilityTx) {
-        let inner = Arc::new(Inner {
-            waker: AtomicWaker::new(),
-            updated: AtomicBool::new(false),
-            available: AtomicBool::new(false),
-            failed: AtomicBool::new(false),
-        });
+        let (mut tx, rx) = broadcast(16);
+        tx.set_overflow(true);
 
         let avail = WorkerAvailability {
-            inner: inner.clone(),
+            notify: rx,
+            available: Arc::new(AtomicBool::new(false)),
+        };
+        let avail_tx = WorkerAvailabilityTx {
+            notify: tx,
+            available: avail.available.clone(),
         };
-        let avail_tx = WorkerAvailabilityTx { inner };
         (avail, avail_tx)
     }
 
-    fn failed(&self) -> bool {
-        self.inner.failed.load(Ordering::Acquire)
-    }
-
     fn available(&self) -> bool {
-        self.inner.available.load(Ordering::Acquire)
-    }
-
-    async fn wait_for_update(&self) {
-        poll_fn(|cx| {
-            if self.inner.updated.load(Ordering::Acquire) {
-                self.inner.updated.store(false, Ordering::Release);
-                Poll::Ready(())
-            } else {
-                self.inner.waker.register(cx.waker());
-                Poll::Pending
-            }
-        })
-        .await;
+        self.available.load(Ordering::Acquire)
     }
 }
 
 impl WorkerAvailabilityTx {
     fn set(&self, val: bool) {
-        let old = self.inner.available.swap(val, Ordering::Release);
-        if old != val {
-            self.inner.updated.store(true, Ordering::Release);
-            self.inner.waker.wake();
+        let old = self.available.swap(val, Ordering::Release);
+        if !old && val {
+            let _ = self.notify.try_broadcast(());
         }
     }
 }
 
-impl Drop for WorkerAvailabilityTx {
-    fn drop(&mut self) {
-        self.inner.failed.store(true, Ordering::Release);
-        self.inner.updated.store(true, Ordering::Release);
-        self.inner.available.store(false, Ordering::Release);
-        self.inner.waker.wake();
-    }
-}
-
 /// Service worker
 ///
 /// Worker accepts message via unbounded channel and starts processing.
 struct WorkerSt<T, F: ServiceFactory<T>> {
     id: WorkerId,
-    rx: Receiver<T>,
+    rx: Pin<Box<dyn Stream<Item = T>>>,
     stop: Pin<Box<dyn Stream<Item = Shutdown>>>,
     factory: F,
     availability: WorkerAvailabilityTx,
@@ -284,43 +253,25 @@ where
     F: ServiceFactory<T> + 'static,
 {
     loop {
-        let mut recv = std::pin::pin!(wrk.rx.recv());
         let fut = poll_fn(|cx| {
-            match svc.poll_ready(cx) {
-                Poll::Ready(Ok(())) => {
-                    wrk.availability.set(true);
-                }
-                Poll::Ready(Err(err)) => {
-                    wrk.availability.set(false);
-                    return Poll::Ready(Err(err));
-                }
-                Poll::Pending => {
-                    wrk.availability.set(false);
-                    return Poll::Pending;
-                }
-            }
+            ready!(svc.poll_ready(cx)?);
 
-            match ready!(recv.as_mut().poll(cx)) {
-                Ok(item) => {
-                    let fut = svc.call(item);
-                    let _ = spawn(async move {
-                        let _ = fut.await;
-                    });
-                    Poll::Ready(Ok::<_, F::Error>(true))
-                }
-                Err(_) => {
-                    log::error!("Server is gone");
-                    Poll::Ready(Ok(false))
-                }
+            if let Some(item) = ready!(Pin::new(&mut wrk.rx).poll_next(cx)) {
+                let fut = svc.call(item);
+                let _ = spawn(async move {
+                    let _ = fut.await;
+                });
             }
+            Poll::Ready(Ok::<(), F::Error>(()))
         });
 
         match select(fut, stream_recv(&mut wrk.stop)).await {
-            Either::Left(Ok(true)) => continue,
+            Either::Left(Ok(())) => continue,
             Either::Left(Err(_)) => {
                 let _ = ntex_rt::spawn(async move {
                     svc.shutdown().await;
                 });
+                wrk.availability.set(false);
             }
             Either::Right(Some(Shutdown { timeout, result })) => {
                 wrk.availability.set(false);
@@ -334,8 +285,7 @@ where
                 stop_svc(wrk.id, svc, timeout, Some(result)).await;
                 return;
             }
-            Either::Left(Ok(false)) | Either::Right(None) => {
-                wrk.availability.set(false);
+            Either::Right(None) => {
                 stop_svc(wrk.id, svc, STOP_TIMEOUT, None).await;
                 return;
             }
@@ -345,6 +295,7 @@ where
         loop {
             match select(wrk.factory.create(()), stream_recv(&mut wrk.stop)).await {
                 Either::Left(Ok(service)) => {
+                    wrk.availability.set(true);
                     svc = Pipeline::new(service).bind();
                     break;
                 }
@@ -385,6 +336,8 @@ where
 {
     availability.set(false);
     let factory = factory?;
+
+    let rx = Box::pin(rx);
     let mut stop = Box::pin(stop);
 
     let svc = match select(factory.create(()), stream_recv(&mut stop)).await {
@@ -403,9 +356,9 @@ where
         svc,
         WorkerSt {
             id,
-            rx,
             factory,
             availability,
+            rx: Box::pin(rx),
             stop: Box::pin(stop),
         },
     ))

ntex-tls/examples/rustls-server.rs

@@ -13,8 +13,9 @@ async fn main() -> io::Result<()> {
     println!("Started openssl echp server: 127.0.0.1:8443");
 
     // load ssl keys
-    let cert_file = &mut BufReader::new(File::open("../examples/cert.pem").unwrap());
-    let key_file = &mut BufReader::new(File::open("../examples/key.pem").unwrap());
+    let cert_file =
+        &mut BufReader::new(File::open("../ntex-tls/examples/cert.pem").unwrap());
+    let key_file = &mut BufReader::new(File::open("../ntex-tls/examples/key.pem").unwrap());
     let keys = rustls_pemfile::private_key(key_file).unwrap().unwrap();
     let cert_chain = rustls_pemfile::certs(cert_file)
         .collect::<Result<Vec<_>, _>>()

ntex-tls/examples/webserver.rs

@@ -8,18 +8,18 @@ use tls_openssl::ssl::{self, SslFiletype, SslMethod};
 
 #[ntex::main]
 async fn main() -> io::Result<()> {
-    std::env::set_var("RUST_LOG", "trace");
-    let _ = env_logger::try_init();
+    //std::env::set_var("RUST_LOG", "trace");
+    //env_logger::init();
 
     println!("Started openssl web server: 127.0.0.1:8443");
 
     // load ssl keys
     let mut builder = ssl::SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
     builder
-        .set_private_key_file("./examples/key.pem", SslFiletype::PEM)
+        .set_private_key_file("../tests/key.pem", SslFiletype::PEM)
         .unwrap();
     builder
-        .set_certificate_chain_file("./examples/cert.pem")
+        .set_certificate_chain_file("../tests/cert.pem")
         .unwrap();
 
     // h2 alpn config

ntex-tls/src/openssl/accept.rs

@@ -128,7 +128,7 @@ impl<F: Filter> Service<Io<F>> for SslAcceptorService {
             };
             let io = io.add_filter(filter);
 
-            log::trace!("Accepting tls connection");
+            log::debug!("Accepting tls connection");
             loop {
                 let result = io.with_buf(|buf| {
                     let filter = io.filter();

ntex/CHANGES.md

@@ -1,23 +1,5 @@
 # Changes
 
-## [2.12.4] - 2025-03-28
-
-* http: Return PayloadError::Incomplete on server disconnect
-
-* web: Expose WebStack for external wrapper support in downstream crates #542
-
-## [2.12.3] - 2025-03-22
-
-* web: Export web::app_service::AppService #534
-
-* http: Add delay for test server availability, could cause connect race
-
-## [2.12.2] - 2025-03-15
-
-* http: Allow to run publish future to completion in case error
-
-* http: Remove brotli support
-
 ## [2.12.1] - 2025-03-14
 
 * Allow to disable test logging (no-test-logging features)

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "2.12.4"
+version = "2.12.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -18,7 +18,7 @@ edition = "2021"
 rust-version = "1.75"
 
 [package.metadata.docs.rs]
-features = ["tokio", "openssl", "rustls", "compress", "cookie", "ws", "ntex-tls/rustls-ring"]
+features = ["tokio", "openssl", "rustls", "compress", "cookie", "ws", "brotli", "ntex-tls/rustls-ring"]
 
 [lib]
 name = "ntex"
@@ -57,6 +57,9 @@ neon-uring = ["ntex-net/neon", "ntex-net/io-uring"]
 # websocket support
 ws = ["dep:sha-1"]
 
+# brotli2 support
+brotli = ["dep:brotli2"]
+
 # disable [ntex::test] logging configuration
 no-test-logging = []
 
@@ -68,11 +71,11 @@ ntex-service = "3.4"
 ntex-macros = "0.1"
 ntex-util = "2.8"
 ntex-bytes = "0.1.27"
-ntex-server = "2.7.4"
+ntex-server = "2.7"
 ntex-h2 = "1.8.6"
 ntex-rt = "0.4.27"
 ntex-io = "2.11"
-ntex-net = "2.5.10"
+ntex-net = "2.5"
 ntex-tls = "2.3"
 
 base64 = "0.22"
@@ -109,12 +112,12 @@ tls-rustls = { version = "0.23", package = "rustls", optional = true, default-fe
 webpki-roots = { version = "0.26", optional = true }
 
 # compression
+brotli2 = { version = "0.3.2", optional = true }
 flate2 = { version = "1.0", optional = true }
 
 [dev-dependencies]
 rand = "0.8"
 time = "0.3"
-oneshot = "0.1"
 futures-util = "0.3"
 tls-openssl = { version = "0.10", package = "openssl" }
 tls-rustls = { version = "0.23", package = "rustls", features = ["ring", "std"], default-features = false }

ntex/src/http/client/h1proto.rs

@@ -1,11 +1,13 @@
-use std::{future::poll_fn, io, io::Write, pin::Pin, task, task::Poll, time::Instant};
+use std::{
+    future::poll_fn, io, io::Write, pin::Pin, task::Context, task::Poll, time::Instant,
+};
 
 use crate::http::body::{BodySize, MessageBody};
 use crate::http::error::PayloadError;
+use crate::http::h1;
 use crate::http::header::{HeaderMap, HeaderValue, HOST};
 use crate::http::message::{RequestHeadType, ResponseHead};
 use crate::http::payload::{Payload, PayloadStream};
-use crate::http::{h1, Version};
 use crate::io::{IoBoxed, RecvError};
 use crate::time::{timeout_checked, Millis};
 use crate::util::{ready, BufMut, Bytes, BytesMut, Stream};
@@ -99,13 +101,7 @@ where
             Ok((head, Payload::None))
         }
         _ => {
-            let pl: PayloadStream = Box::pin(PlStream::new(
-                io,
-                codec,
-                created,
-                pool,
-                head.version == Version::HTTP_10,
-            ));
+            let pl: PayloadStream = Box::pin(PlStream::new(io, codec, created, pool));
             Ok((head, pl.into()))
         }
     }
@@ -141,7 +137,6 @@ pub(super) struct PlStream {
     io: Option<IoBoxed>,
     codec: h1::ClientPayloadCodec,
     created: Instant,
-    http_10: bool,
     pool: Option<Acquired>,
 }
 
@@ -151,14 +146,12 @@ impl PlStream {
         codec: h1::ClientCodec,
         created: Instant,
         pool: Option<Acquired>,
-        http_10: bool,
     ) -> Self {
         PlStream {
             io: Some(io),
             codec: codec.into_payload_codec(),
             created,
             pool,
-            http_10,
         }
     }
 }
@@ -168,46 +161,41 @@ impl Stream for PlStream {
 
     fn poll_next(
         mut self: Pin<&mut Self>,
-        cx: &mut task::Context<'_>,
+        cx: &mut Context<'_>,
     ) -> Poll<Option<Self::Item>> {
         let mut this = self.as_mut();
         loop {
-            let item = ready!(this.io.as_ref().unwrap().poll_recv(&this.codec, cx));
-            return Poll::Ready(Some(match item {
-                Ok(chunk) => {
-                    if let Some(chunk) = chunk {
-                        Ok(chunk)
-                    } else {
-                        release_connection(
-                            this.io.take().unwrap(),
-                            !this.codec.keepalive(),
-                            this.created,
-                            this.pool.take(),
-                        );
-                        return Poll::Ready(None);
+            return Poll::Ready(Some(
+                match ready!(this.io.as_ref().unwrap().poll_recv(&this.codec, cx)) {
+                    Ok(chunk) => {
+                        if let Some(chunk) = chunk {
+                            Ok(chunk)
+                        } else {
+                            release_connection(
+                                this.io.take().unwrap(),
+                                !this.codec.keepalive(),
+                                this.created,
+                                this.pool.take(),
+                            );
+                            return Poll::Ready(None);
+                        }
                     }
-                }
-                Err(RecvError::KeepAlive) => {
-                    Err(io::Error::new(io::ErrorKind::TimedOut, "Keep-alive").into())
-                }
-                Err(RecvError::Stop) => {
-                    Err(io::Error::new(io::ErrorKind::Other, "Dispatcher stopped").into())
-                }
-                Err(RecvError::WriteBackpressure) => {
-                    ready!(this.io.as_ref().unwrap().poll_flush(cx, false))?;
-                    continue;
-                }
-                Err(RecvError::Decoder(err)) => Err(err),
-                Err(RecvError::PeerGone(Some(err))) => {
-                    Err(PayloadError::Incomplete(Some(err)))
-                }
-                Err(RecvError::PeerGone(None)) => {
-                    if this.http_10 {
-                        return Poll::Ready(None);
+                    Err(RecvError::KeepAlive) => {
+                        Err(io::Error::new(io::ErrorKind::TimedOut, "Keep-alive").into())
                     }
-                    Err(PayloadError::Incomplete(None))
-                }
-            }));
+                    Err(RecvError::Stop) => {
+                        Err(io::Error::new(io::ErrorKind::Other, "Dispatcher stopped")
+                            .into())
+                    }
+                    Err(RecvError::WriteBackpressure) => {
+                        ready!(this.io.as_ref().unwrap().poll_flush(cx, false))?;
+                        continue;
+                    }
+                    Err(RecvError::Decoder(err)) => Err(err),
+                    Err(RecvError::PeerGone(Some(err))) => Err(err.into()),
+                    Err(RecvError::PeerGone(None)) => return Poll::Ready(None),
+                },
+            ));
         }
     }
 }

ntex/src/http/client/response.rs

@@ -387,8 +387,8 @@ impl Future for ReadBody {
         let this = self.get_mut();
 
         loop {
-            return match Pin::new(&mut this.stream).poll_next(cx) {
-                Poll::Ready(Some(Ok(chunk))) => {
+            return match Pin::new(&mut this.stream).poll_next(cx)? {
+                Poll::Ready(Some(chunk)) => {
                     if this.limit > 0 && (this.buf.len() + chunk.len()) > this.limit {
                         Poll::Ready(Err(PayloadError::Overflow))
                     } else {
@@ -397,7 +397,6 @@ impl Future for ReadBody {
                     }
                 }
                 Poll::Ready(None) => Poll::Ready(Ok(this.buf.split().freeze())),
-                Poll::Ready(Some(Err(err))) => Poll::Ready(Err(err)),
                 Poll::Pending => {
                     if this.timeout.poll_elapsed(cx).is_ready() {
                         Poll::Ready(Err(PayloadError::Incomplete(Some(

ntex/src/http/encoding/decoder.rs

@@ -1,5 +1,7 @@
 use std::{future::Future, io, io::Write, pin::Pin, task::Context, task::Poll};
 
+#[cfg(feature = "brotli")]
+use brotli2::write::BrotliDecoder;
 use flate2::write::{GzDecoder, ZlibDecoder};
 
 use super::Writer;
@@ -25,6 +27,10 @@ where
     #[inline]
     pub fn new(stream: S, encoding: ContentEncoding) -> Decoder<S> {
         let decoder = match encoding {
+            #[cfg(feature = "brotli")]
+            ContentEncoding::Br => Some(ContentDecoder::Br(Box::new(BrotliDecoder::new(
+                Writer::new(),
+            )))),
             ContentEncoding::Deflate => Some(ContentDecoder::Deflate(Box::new(
                 ZlibDecoder::new(Writer::new()),
             ))),
@@ -131,11 +137,25 @@ where
 enum ContentDecoder {
     Deflate(Box<ZlibDecoder<Writer>>),
     Gzip(Box<GzDecoder<Writer>>),
+    #[cfg(feature = "brotli")]
+    Br(Box<BrotliDecoder<Writer>>),
 }
 
 impl ContentDecoder {
     fn feed_eof(&mut self) -> io::Result<Option<Bytes>> {
         match self {
+            #[cfg(feature = "brotli")]
+            ContentDecoder::Br(ref mut decoder) => match decoder.flush() {
+                Ok(()) => {
+                    let b = decoder.get_mut().take();
+                    if !b.is_empty() {
+                        Ok(Some(b))
+                    } else {
+                        Ok(None)
+                    }
+                }
+                Err(e) => Err(e),
+            },
             ContentDecoder::Gzip(ref mut decoder) => match decoder.try_finish() {
                 Ok(_) => {
                     let b = decoder.get_mut().take();
@@ -163,6 +183,19 @@ impl ContentDecoder {
 
     fn feed_data(&mut self, data: Bytes) -> io::Result<Option<Bytes>> {
         match self {
+            #[cfg(feature = "brotli")]
+            ContentDecoder::Br(ref mut decoder) => match decoder.write_all(&data) {
+                Ok(_) => {
+                    decoder.flush()?;
+                    let b = decoder.get_mut().take();
+                    if !b.is_empty() {
+                        Ok(Some(b))
+                    } else {
+                        Ok(None)
+                    }
+                }
+                Err(e) => Err(e),
+            },
             ContentDecoder::Gzip(ref mut decoder) => match decoder.write_all(&data) {
                 Ok(_) => {
                     decoder.flush()?;

ntex/src/http/encoding/encoder.rs

@@ -1,6 +1,8 @@
 //! Stream encoder
 use std::{fmt, future::Future, io, io::Write, pin::Pin, task::Context, task::Poll};
 
+#[cfg(feature = "brotli")]
+use brotli2::write::BrotliEncoder;
 use flate2::write::{GzEncoder, ZlibEncoder};
 
 use crate::http::body::{Body, BodySize, MessageBody, ResponseBody};
@@ -189,11 +191,23 @@ fn update_head(encoding: ContentEncoding, head: &mut ResponseHead) {
 enum ContentEncoder {
     Deflate(ZlibEncoder<Writer>),
     Gzip(GzEncoder<Writer>),
+    #[cfg(feature = "brotli")]
+    Br(BrotliEncoder<Writer>),
 }
 
 impl ContentEncoder {
     fn can_encode(encoding: ContentEncoding) -> bool {
-        matches!(encoding, ContentEncoding::Deflate | ContentEncoding::Gzip)
+        #[cfg(feature = "brotli")]
+        {
+            matches!(
+                encoding,
+                ContentEncoding::Deflate | ContentEncoding::Gzip | ContentEncoding::Br
+            )
+        }
+        #[cfg(not(feature = "brotli"))]
+        {
+            matches!(encoding, ContentEncoding::Deflate | ContentEncoding::Gzip)
+        }
     }
 
     fn encoder(encoding: ContentEncoding) -> Option<Self> {
@@ -206,12 +220,18 @@ impl ContentEncoder {
                 Writer::new(),
                 flate2::Compression::fast(),
             ))),
+            #[cfg(feature = "brotli")]
+            ContentEncoding::Br => {
+                Some(ContentEncoder::Br(BrotliEncoder::new(Writer::new(), 3)))
+            }
             _ => None,
         }
     }
 
     fn take(&mut self) -> Bytes {
         match *self {
+            #[cfg(feature = "brotli")]
+            ContentEncoder::Br(ref mut encoder) => encoder.get_mut().take(),
             ContentEncoder::Deflate(ref mut encoder) => encoder.get_mut().take(),
             ContentEncoder::Gzip(ref mut encoder) => encoder.get_mut().take(),
         }
@@ -219,6 +239,11 @@ impl ContentEncoder {
 
     fn finish(self) -> Result<Bytes, io::Error> {
         match self {
+            #[cfg(feature = "brotli")]
+            ContentEncoder::Br(encoder) => match encoder.finish() {
+                Ok(writer) => Ok(writer.buf.freeze()),
+                Err(err) => Err(err),
+            },
             ContentEncoder::Gzip(encoder) => match encoder.finish() {
                 Ok(writer) => Ok(writer.buf.freeze()),
                 Err(err) => Err(err),
@@ -232,6 +257,14 @@ impl ContentEncoder {
 
     fn write(&mut self, data: &[u8]) -> Result<(), io::Error> {
         match *self {
+            #[cfg(feature = "brotli")]
+            ContentEncoder::Br(ref mut encoder) => match encoder.write_all(data) {
+                Ok(_) => Ok(()),
+                Err(err) => {
+                    log::trace!("Error decoding br encoding: {}", err);
+                    Err(err)
+                }
+            },
             ContentEncoder::Gzip(ref mut encoder) => match encoder.write_all(data) {
                 Ok(_) => Ok(()),
                 Err(err) => {
@@ -255,6 +288,8 @@ impl fmt::Debug for ContentEncoder {
         match self {
             ContentEncoder::Deflate(_) => write!(f, "ContentEncoder::Deflate"),
             ContentEncoder::Gzip(_) => write!(f, "ContentEncoder::Gzip"),
+            #[cfg(feature = "brotli")]
+            ContentEncoder::Br(_) => write!(f, "ContentEncoder::Br"),
         }
     }
 }

ntex/src/http/h1/decoder.rs

@@ -107,12 +107,12 @@ pub(super) trait MessageType: Sized {
                 };
                 match name {
                     header::CONTENT_LENGTH if content_length.is_some() || chunked => {
-                        log::trace!("multiple Content-Length not allowed");
+                        log::debug!("multiple Content-Length not allowed");
                         return Err(DecodeError::Header);
                     }
                     header::CONTENT_LENGTH => match value.to_str() {
                         Ok(s) if s.trim_start().starts_with('+') => {
-                            log::trace!("illegal Content-Length: {:?}", s);
+                            log::debug!("illegal Content-Length: {:?}", s);
                             return Err(DecodeError::Header);
                         }
                         Ok(s) => {
@@ -121,18 +121,18 @@ pub(super) trait MessageType: Sized {
                                 // headers have been processed to prevent request smuggling issues
                                 content_length = Some(len);
                             } else {
-                                log::trace!("illegal Content-Length: {:?}", s);
+                                log::debug!("illegal Content-Length: {:?}", s);
                                 return Err(DecodeError::Header);
                             }
                         }
                         Err(_) => {
-                            log::trace!("illegal Content-Length: {:?}", value);
+                            log::debug!("illegal Content-Length: {:?}", value);
                             return Err(DecodeError::Header);
                         }
                     },
                     // transfer-encoding
                     header::TRANSFER_ENCODING if seen_te => {
-                        log::trace!("Transfer-Encoding header usage is not allowed");
+                        log::debug!("Transfer-Encoding header usage is not allowed");
                         return Err(DecodeError::Header);
                     }
                     header::TRANSFER_ENCODING if version == Version::HTTP_11 => {
@@ -144,7 +144,7 @@ pub(super) trait MessageType: Sized {
                             } else if s.eq_ignore_ascii_case("identity") {
                                 // allow silently since multiple TE headers are already checked
                             } else {
-                                log::trace!("illegal Transfer-Encoding: {:?}", s);
+                                log::debug!("illegal Transfer-Encoding: {:?}", s);
                                 return Err(DecodeError::Header);
                             }
                         } else {
@@ -266,7 +266,7 @@ impl MessageType for Request {
         // disallow HTTP/1.0 POST requests that do not contain a Content-Length headers
         // see https://datatracker.ietf.org/doc/html/rfc1945#section-7.2.2
         if ver == Version::HTTP_10 && method == Method::POST && length.is_none() {
-            log::trace!("no Content-Length specified for HTTP/1.0 POST request");
+            log::debug!("no Content-Length specified for HTTP/1.0 POST request");
             return Err(DecodeError::Header);
         }
 
@@ -682,7 +682,7 @@ impl ChunkedState {
                 Poll::Ready(Ok(ChunkedState::Size))
             }
             None => {
-                log::trace!("chunk size would overflow u64");
+                log::debug!("chunk size would overflow u64");
                 Poll::Ready(Err(DecodeError::InvalidInput(
                     "Invalid chunk size line: Size is too big",
                 )))

ntex/src/http/h1/dispatcher.rs

@@ -1,5 +1,5 @@
 //! HTTP/1 protocol dispatcher
-use std::{error, future, io, marker, mem, pin::Pin, rc::Rc, task::Context, task::Poll};
+use std::{error, future, io, marker, pin::Pin, rc::Rc, task::Context, task::Poll};
 
 use crate::io::{Decoded, Filter, Io, IoStatusUpdate, RecvError};
 use crate::service::{PipelineCall, Service};
@@ -144,20 +144,7 @@ where
                         inner.send_response(res, body)
                     }
                     Poll::Ready(Err(err)) => inner.control(Control::err(err)),
-                    Poll::Pending => {
-                        // state changed because of error.
-                        // spawn current publish future to runtime
-                        // so it could complete error handling
-                        let st = ready!(inner.poll_request(cx));
-                        if inner.payload.is_some() {
-                            if let State::CallPublish { fut } =
-                                mem::replace(&mut *this.st, State::ReadRequest)
-                            {
-                                crate::rt::spawn(fut);
-                            }
-                        }
-                        st
-                    }
+                    Poll::Pending => ready!(inner.poll_request(cx)),
                 },
                 // handle control service responses
                 State::CallControl { fut } => match Pin::new(fut).poll(cx) {
@@ -352,7 +339,7 @@ where
                 .io
                 .encode(Message::Item((msg, body.size())), &self.codec)
                 .map_err(|err| {
-                    if let Some(ref mut payload) = self.payload {
+                    if let Some(mut payload) = self.payload.take() {
                         payload.1.set_error(PayloadError::Incomplete(None));
                     }
                     err
@@ -451,7 +438,7 @@ where
     }
 
     fn set_payload_error(&mut self, err: PayloadError) {
-        if let Some(ref mut payload) = self.payload {
+        if let Some(mut payload) = self.payload.take() {
             payload.1.set_error(err);
         }
     }

ntex/src/http/h1/payload.rs

@@ -3,7 +3,8 @@ use std::rc::{Rc, Weak};
 use std::task::{Context, Poll};
 use std::{cell::RefCell, collections::VecDeque, pin::Pin};
 
-use crate::{http::error::PayloadError, task::LocalWaker, util::Bytes, util::Stream};
+use crate::http::error::PayloadError;
+use crate::{task::LocalWaker, util::Bytes, util::Stream};
 
 /// max buffer size 32k
 const MAX_BUFFER_SIZE: usize = 32_768;
@@ -118,7 +119,7 @@ impl PayloadSender {
         // we check only if Payload (other side) is alive,
         // otherwise always return true (consume payload)
         if let Some(shared) = self.inner.upgrade() {
-            if shared.borrow().flags.contains(Flags::NEED_READ) {
+            if shared.borrow().need_read {
                 PayloadStatus::Read
             } else {
                 shared.borrow_mut().io_task.register(cx.waker());
@@ -130,20 +131,12 @@ impl PayloadSender {
     }
 }
 
-bitflags::bitflags! {
-    #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
-    struct Flags: u8 {
-        const EOF       = 0b0000_0001;
-        const ERROR     = 0b0000_0010;
-        const NEED_READ = 0b0000_0100;
-    }
-}
-
 #[derive(Debug)]
 struct Inner {
     len: usize,
-    flags: Flags,
+    eof: bool,
     err: Option<PayloadError>,
+    need_read: bool,
     items: VecDeque<Bytes>,
     task: LocalWaker,
     io_task: LocalWaker,
@@ -151,16 +144,12 @@ struct Inner {
 
 impl Inner {
     fn new(eof: bool) -> Self {
-        let flags = if eof {
-            Flags::EOF | Flags::NEED_READ
-        } else {
-            Flags::NEED_READ
-        };
         Inner {
-            flags,
+            eof,
             len: 0,
             err: None,
             items: VecDeque::new(),
+            need_read: true,
             task: LocalWaker::new(),
             io_task: LocalWaker::new(),
         }
@@ -168,23 +157,18 @@ impl Inner {
 
     fn set_error(&mut self, err: PayloadError) {
         self.err = Some(err);
-        self.flags.insert(Flags::ERROR);
         self.task.wake()
     }
 
     fn feed_eof(&mut self) {
-        self.flags.insert(Flags::EOF);
+        self.eof = true;
         self.task.wake()
     }
 
     fn feed_data(&mut self, data: Bytes) {
         self.len += data.len();
         self.items.push_back(data);
-        if self.len < MAX_BUFFER_SIZE {
-            self.flags.insert(Flags::NEED_READ);
-        } else {
-            self.flags.remove(Flags::NEED_READ);
-        }
+        self.need_read = self.len < MAX_BUFFER_SIZE;
         self.task.wake();
     }
 
@@ -194,25 +178,19 @@ impl Inner {
     ) -> Poll<Option<Result<Bytes, PayloadError>>> {
         if let Some(data) = self.items.pop_front() {
             self.len -= data.len();
-            if self.len < MAX_BUFFER_SIZE {
-                self.flags.insert(Flags::NEED_READ);
-            } else {
-                self.flags.remove(Flags::NEED_READ);
-            }
+            self.need_read = self.len < MAX_BUFFER_SIZE;
 
-            if self.flags.contains(Flags::NEED_READ)
-                && !self.flags.intersects(Flags::EOF | Flags::ERROR)
-            {
+            if self.need_read && !self.eof {
                 self.task.register(cx.waker());
             }
             self.io_task.wake();
             Poll::Ready(Some(Ok(data)))
         } else if let Some(err) = self.err.take() {
             Poll::Ready(Some(Err(err)))
-        } else if self.flags.intersects(Flags::EOF | Flags::ERROR) {
+        } else if self.eof {
             Poll::Ready(None)
         } else {
-            self.flags.insert(Flags::NEED_READ);
+            self.need_read = true;
             self.task.register(cx.waker());
             self.io_task.wake();
             Poll::Pending

ntex/src/http/test.rs

@@ -11,7 +11,7 @@ use crate::server::Server;
 use crate::service::ServiceFactory;
 #[cfg(feature = "ws")]
 use crate::ws::{error::WsClientError, WsClient, WsConnection};
-use crate::{rt::System, time::sleep, time::Millis, time::Seconds, util::Bytes};
+use crate::{rt::System, time::Millis, time::Seconds, util::Bytes};
 
 use super::client::{Client, ClientRequest, ClientResponse, Connector};
 use super::error::{HttpError, PayloadError};
@@ -244,14 +244,15 @@ where
                 .workers(1)
                 .disable_signals()
                 .run();
-
-            crate::rt::spawn(async move {
-                sleep(Millis(125)).await;
-                tx.send((system, srv, local_addr)).unwrap();
-            });
+            tx.send((system, srv, local_addr)).unwrap();
             Ok(())
         })
     });
+    // wait for server
+    if std::env::var("GITHUB_ACTIONS") == Ok("true".to_string()) {
+        thread::sleep(std::time::Duration::from_millis(150));
+    }
+
     let (system, server, addr) = rx.recv().unwrap();
 
     TestServer {

ntex/src/web/config.rs

@@ -68,7 +68,7 @@ pub struct ServiceConfig<Err = DefaultError> {
 }
 
 impl<Err: ErrorRenderer> ServiceConfig<Err> {
-    pub fn new() -> Self {
+    pub(crate) fn new() -> Self {
         Self {
             services: Vec::new(),
             state: Extensions::new(),
@@ -132,7 +132,7 @@ mod tests {
     use crate::http::{Method, StatusCode};
     use crate::util::Bytes;
     use crate::web::test::{call_service, init_service, read_body, TestRequest};
-    use crate::web::{self, App, DefaultError, HttpRequest, HttpResponse};
+    use crate::web::{self, App, HttpRequest, HttpResponse};
 
     #[crate::rt_test]
     async fn test_configure_state() {
@@ -205,11 +205,4 @@ mod tests {
         let resp = call_service(&srv, req).await;
         assert_eq!(resp.status(), StatusCode::OK);
     }
-
-    #[test]
-    fn test_new_service_config() {
-        let cfg: ServiceConfig<DefaultError> = ServiceConfig::new();
-        assert!(cfg.services.is_empty());
-        assert!(cfg.external.is_empty());
-    }
 }

ntex/src/web/mod.rs

@@ -82,7 +82,7 @@ mod route;
 mod scope;
 mod server;
 mod service;
-pub mod stack;
+mod stack;
 pub mod test;
 pub mod types;
 mod util;
@@ -128,7 +128,6 @@ pub mod dev {
     //! The purpose of this module is to alleviate imports of many common
     //! traits by adding a glob import to the top of ntex::web heavy modules:
 
-    pub use crate::web::app_service::AppService;
     pub use crate::web::config::AppConfig;
     pub use crate::web::info::ConnectionInfo;
     pub use crate::web::rmap::ResourceMap;

ntex/src/web/test.rs

@@ -697,13 +697,14 @@ where
             .set_tag("test", "WEB-SRV")
             .run();
 
-            crate::rt::spawn(async move {
-                sleep(Millis(125)).await;
-                tx.send((System::current(), srv, local_addr)).unwrap();
-            });
+            tx.send((System::current(), srv, local_addr)).unwrap();
             Ok(())
         })
     });
+    // wait for server
+    if std::env::var("GITHUB_ACTIONS") == Ok("true".to_string()) {
+        thread::sleep(std::time::Duration::from_millis(150));
+    }
 
     let (system, server, addr) = rx.recv().unwrap();
 

ntex/src/ws/frame.rs

@@ -116,7 +116,7 @@ impl Parser {
                 return Err(ProtocolError::InvalidLength(length));
             }
             OpCode::Close if length > 125 => {
-                log::trace!("Received close frame with payload length exceeding 125. Morphing to protocol close frame.");
+                log::debug!("Received close frame with payload length exceeding 125. Morphing to protocol close frame.");
                 return Ok(Some((true, OpCode::Close, None)));
             }
             _ => (),

ntex/tests/connect.rs

@@ -1,8 +1,9 @@
 use std::{io, rc::Rc};
 
+use ntex::codec::BytesCodec;
+use ntex::connect::Connect;
 use ntex::io::{types::PeerAddr, Io};
 use ntex::service::{chain_factory, fn_service, Pipeline, ServiceFactory};
-use ntex::{codec::BytesCodec, connect::Connect};
 use ntex::{server::build_test_server, server::test_server, time, util::Bytes};
 
 #[cfg(feature = "rustls")]

ntex/tests/http_awc_client.rs

@@ -3,14 +3,16 @@ use std::io::{Read, Write};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
+use brotli2::write::BrotliEncoder;
 use coo_kie::Cookie;
 use flate2::{read::GzDecoder, write::GzEncoder, write::ZlibEncoder, Compression};
+use futures_util::stream::once;
 use rand::Rng;
 
-use ntex::http::client::error::SendRequestError;
+use ntex::http::client::error::{JsonPayloadError, SendRequestError};
 use ntex::http::client::{Client, Connector};
 use ntex::http::test::server as test_server;
-use ntex::http::{header, HttpMessage, HttpService};
+use ntex::http::{header, HttpMessage, HttpService, Method};
 use ntex::service::{chain_factory, map_config};
 use ntex::web::dev::AppConfig;
 use ntex::web::middleware::Compress;
@@ -508,21 +510,19 @@ async fn test_client_gzip_encoding_large() {
 async fn test_client_gzip_encoding_large_random() {
     let data = rand::thread_rng()
         .sample_iter(&rand::distributions::Alphanumeric)
-        .take(1_048_500)
+        .take(100_000)
         .map(char::from)
         .collect::<String>();
 
     let srv = test::server(|| {
-        App::new()
-            .state(web::types::PayloadConfig::default().limit(1_048_576))
-            .service(web::resource("/").route(web::to(|data: Bytes| async move {
-                let mut e = GzEncoder::new(Vec::new(), Compression::default());
-                e.write_all(&data).unwrap();
-                let data = e.finish().unwrap();
-                HttpResponse::Ok()
-                    .header("content-encoding", "gzip")
-                    .body(data)
-            })))
+        App::new().service(web::resource("/").route(web::to(|data: Bytes| async move {
+            let mut e = GzEncoder::new(Vec::new(), Compression::default());
+            e.write_all(&data).unwrap();
+            let data = e.finish().unwrap();
+            HttpResponse::Ok()
+                .header("content-encoding", "gzip")
+                .body(data)
+        })))
     });
 
     // client request
@@ -530,10 +530,130 @@ async fn test_client_gzip_encoding_large_random() {
     assert!(response.status().is_success());
 
     // read response
-    let bytes = response.body().limit(1_048_576).await.unwrap();
+    let bytes = response.body().await.unwrap();
     assert_eq!(bytes, Bytes::from(data));
 }
 
+#[ntex::test]
+async fn test_client_brotli_encoding() {
+    let srv = test::server(|| {
+        App::new().service(web::resource("/").route(web::to(|data: Bytes| async move {
+            let mut e = BrotliEncoder::new(Vec::new(), 5);
+            e.write_all(&data).unwrap();
+            let data = e.finish().unwrap();
+            HttpResponse::Ok()
+                .header("content-encoding", "br")
+                .body(data)
+        })))
+    });
+
+    // client request
+    let mut response = srv.post("/").send_body(STR).await.unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
+}
+
+#[ntex::test]
+async fn test_client_brotli_encoding_large_random() {
+    let data = rand::thread_rng()
+        .sample_iter(&rand::distributions::Alphanumeric)
+        .take(70_000)
+        .map(char::from)
+        .collect::<String>();
+
+    let srv = test::server(|| {
+        App::new().service(web::resource("/").route(web::to(|data: Bytes| async move {
+            let mut e = BrotliEncoder::new(Vec::new(), 5);
+            e.write_all(&data).unwrap();
+            let data = e.finish().unwrap();
+            HttpResponse::Ok()
+                .header("content-encoding", "br")
+                .body(data)
+        })))
+    });
+
+    // client request
+    let mut response = srv.post("/").send_body(data.clone()).await.unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes.len(), data.len());
+    assert_eq!(bytes, Bytes::from(data.clone()));
+
+    // frozen request
+    let request = srv.post("/").timeout(Seconds(30)).freeze().unwrap();
+    assert_eq!(request.get_method(), Method::POST);
+    assert_eq!(request.get_uri(), srv.url("/").as_str());
+    let mut response = request.send_body(data.clone()).await.unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes.len(), data.len());
+    assert_eq!(bytes, Bytes::from(data.clone()));
+
+    // extra header
+    let mut response = request
+        .extra_header("x-test2", "222")
+        .send_body(data.clone())
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes.len(), data.len());
+    assert_eq!(bytes, Bytes::from(data.clone()));
+
+    // client stream request
+    let mut response = srv
+        .post("/")
+        .send_stream(once(Ready::Ok::<_, JsonPayloadError>(Bytes::from(
+            data.clone(),
+        ))))
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes.len(), data.len());
+    assert_eq!(bytes, Bytes::from(data.clone()));
+
+    // frozen request
+    let request = srv.post("/").timeout(Seconds(30)).freeze().unwrap();
+    let mut response = request
+        .send_stream(once(Ready::Ok::<_, JsonPayloadError>(Bytes::from(
+            data.clone(),
+        ))))
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes.len(), data.len());
+    assert_eq!(bytes, Bytes::from(data.clone()));
+
+    let mut response = request
+        .extra_header("x-test2", "222")
+        .send_stream(once(Ready::Ok::<_, JsonPayloadError>(Bytes::from(
+            data.clone(),
+        ))))
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes.len(), data.len());
+    assert_eq!(bytes, Bytes::from(data.clone()));
+}
+
 #[ntex::test]
 async fn test_client_deflate_encoding() {
     let srv = test::server(|| {
@@ -682,18 +802,15 @@ async fn client_read_until_eof() {
         for stream in lst.incoming() {
             if let Ok(mut stream) = stream {
                 let mut b = [0; 1000];
-                log::debug!("Reading request");
-                let res = stream.read(&mut b).unwrap();
-                log::debug!("Read {:?}", res);
-                let res = stream
+                let _ = stream.read(&mut b).unwrap();
+                let _ = stream
                     .write_all(b"HTTP/1.0 200 OK\r\nconnection: close\r\n\r\nwelcome!");
-                log::debug!("Sent {:?}", res);
             } else {
                 break;
             }
         }
     });
-    sleep(Millis(500)).await;
+    sleep(Millis(300)).await;
 
     // client request
     let req = Client::build()

ntex/tests/http_openssl.rs

@@ -1,6 +1,5 @@
 #![cfg(feature = "openssl")]
-use std::io;
-use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc, Mutex};
+use std::{io, sync::atomic::AtomicUsize, sync::atomic::Ordering, sync::Arc};
 
 use futures_util::stream::{once, Stream, StreamExt};
 use tls_openssl::ssl::{AlpnError, SslAcceptor, SslFiletype, SslMethod};
@@ -425,12 +424,11 @@ async fn test_h2_service_error() {
     assert_eq!(bytes, Bytes::from_static(b"error"));
 }
 
-struct SetOnDrop(Arc<AtomicUsize>, Arc<Mutex<Option<::oneshot::Sender<()>>>>);
+struct SetOnDrop(Arc<AtomicUsize>);
 
 impl Drop for SetOnDrop {
     fn drop(&mut self) {
         self.0.fetch_add(1, Ordering::Relaxed);
-        let _ = self.1.lock().unwrap().take().unwrap().send(());
     }
 }
 
@@ -438,20 +436,17 @@ impl Drop for SetOnDrop {
 async fn test_h2_client_drop() -> io::Result<()> {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
-    let (tx, rx) = ::oneshot::channel();
-    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
-        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build()
             .h2(move |req: Request| {
-                let st = SetOnDrop(count.clone(), tx.clone());
+                let count = count.clone();
                 async move {
+                    let _st = SetOnDrop(count);
                     assert!(req.peer_addr().is_some());
                     assert_eq!(req.version(), Version::HTTP_2);
-                    sleep(Seconds(30)).await;
-                    drop(st);
+                    sleep(Seconds(100)).await;
                     Ok::<_, io::Error>(Response::Ok().finish())
                 }
             })
@@ -459,9 +454,9 @@ async fn test_h2_client_drop() -> io::Result<()> {
             .map_err(|_| ())
     });
 
-    let result = timeout(Millis(1500), srv.srequest(Method::GET, "/").send()).await;
+    let result = timeout(Millis(250), srv.srequest(Method::GET, "/").send()).await;
     assert!(result.is_err());
-    let _ = timeout(Millis(1500), rx).await;
+    sleep(Millis(150)).await;
     assert_eq!(count.load(Ordering::Relaxed), 1);
     Ok(())
 }
@@ -544,19 +539,13 @@ async fn test_ws_transport() {
 async fn test_h2_graceful_shutdown() -> io::Result<()> {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
-    let (tx, rx) = ::oneshot::channel();
-    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
-        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build()
             .h2(move |_| {
                 let count = count.clone();
                 count.fetch_add(1, Ordering::Relaxed);
-                if count.load(Ordering::Relaxed) == 2 {
-                    let _ = tx.lock().unwrap().take().unwrap().send(());
-                }
                 async move {
                     sleep(Millis(1000)).await;
                     count.fetch_sub(1, Ordering::Relaxed);
@@ -577,7 +566,7 @@ async fn test_h2_graceful_shutdown() -> io::Result<()> {
         let _ = req.send().await.unwrap();
         sleep(Millis(100000)).await;
     });
-    let _ = rx.await;
+    sleep(Millis(150)).await;
     assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let (tx, rx) = oneshot::channel();
@@ -585,6 +574,8 @@ async fn test_h2_graceful_shutdown() -> io::Result<()> {
         srv.stop().await;
         let _ = tx.send(());
     });
+    sleep(Millis(150)).await;
+    assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 0);

ntex/tests/http_server.rs

@@ -1,4 +1,4 @@
-use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc, Mutex};
+use std::sync::{atomic::AtomicUsize, atomic::Ordering, Arc};
 use std::{io, io::Read, io::Write, net};
 
 use futures_util::future::{self, FutureExt};
@@ -405,36 +405,6 @@ async fn test_http1_handle_not_consumed_payload() {
     assert_eq!(&data[..17], b"HTTP/1.1 200 OK\r\n");
 }
 
-/// Handle payload errors (keep-alive, disconnects)
-#[ntex::test]
-async fn test_http1_handle_payload_errors() {
-    let count = Arc::new(AtomicUsize::new(0));
-    let count2 = count.clone();
-
-    let srv = test_server(move || {
-        let count = count2.clone();
-        HttpService::build().h1(move |mut req: Request| {
-            let count = count.clone();
-            async move {
-                let mut pl = req.take_payload();
-                let result = pl.recv().await;
-                if result.unwrap().is_err() {
-                    count.fetch_add(1, Ordering::Relaxed);
-                }
-                Ok::<_, io::Error>(Response::Ok().finish())
-            }
-        })
-    });
-
-    let mut stream = net::TcpStream::connect(srv.addr()).unwrap();
-    let _ =
-        stream.write_all(b"GET /test/tests/test HTTP/1.1\r\ncontent-length: 99999\r\n\r\n");
-    sleep(Millis(250)).await;
-    drop(stream);
-    sleep(Millis(250)).await;
-    assert_eq!(count.load(Ordering::Acquire), 1);
-}
-
 #[ntex::test]
 async fn test_content_length() {
     let srv = test_server(|| {
@@ -723,12 +693,11 @@ async fn test_h1_service_error() {
     assert_eq!(bytes, Bytes::from_static(b"error"));
 }
 
-struct SetOnDrop(Arc<AtomicUsize>, Option<::oneshot::Sender<()>>);
+struct SetOnDrop(Arc<AtomicUsize>);
 
 impl Drop for SetOnDrop {
     fn drop(&mut self) {
         self.0.fetch_add(1, Ordering::Relaxed);
-        let _ = self.1.take().unwrap().send(());
     }
 }
 
@@ -736,28 +705,24 @@ impl Drop for SetOnDrop {
 async fn test_h1_client_drop() -> io::Result<()> {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
-    let (tx, rx) = ::oneshot::channel();
-    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
-        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build().h1(move |req: Request| {
-            let tx = tx.clone();
             let count = count.clone();
             async move {
-                let _st = SetOnDrop(count, tx.lock().unwrap().take());
+                let _st = SetOnDrop(count);
                 assert!(req.peer_addr().is_some());
                 assert_eq!(req.version(), Version::HTTP_11);
-                sleep(Millis(50000)).await;
+                sleep(Seconds(100)).await;
                 Ok::<_, io::Error>(Response::Ok().finish())
             }
         })
     });
 
-    let result = timeout(Millis(1500), srv.request(Method::GET, "/").send()).await;
+    let result = timeout(Millis(100), srv.request(Method::GET, "/").send()).await;
     assert!(result.is_err());
-    let _ = rx.await;
+    sleep(Millis(250)).await;
     assert_eq!(count.load(Ordering::Relaxed), 1);
     Ok(())
 }
@@ -766,18 +731,12 @@ async fn test_h1_client_drop() -> io::Result<()> {
 async fn test_h1_gracefull_shutdown() {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
-    let (tx, rx) = ::oneshot::channel();
-    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
-        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build().h1(move |_: Request| {
             let count = count.clone();
             count.fetch_add(1, Ordering::Relaxed);
-            if count.load(Ordering::Relaxed) == 2 {
-                let _ = tx.lock().unwrap().take().unwrap().send(());
-            }
             async move {
                 sleep(Millis(1000)).await;
                 count.fetch_sub(1, Ordering::Relaxed);
@@ -792,7 +751,7 @@ async fn test_h1_gracefull_shutdown() {
     let mut stream2 = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream2.write_all(b"GET /index.html HTTP/1.1\r\n\r\n");
 
-    let _ = rx.await;
+    sleep(Millis(150)).await;
     assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let (tx, rx) = oneshot::channel();
@@ -800,6 +759,8 @@ async fn test_h1_gracefull_shutdown() {
         srv.stop().await;
         let _ = tx.send(());
     });
+    sleep(Millis(150)).await;
+    assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 0);
@@ -809,18 +770,12 @@ async fn test_h1_gracefull_shutdown() {
 async fn test_h1_gracefull_shutdown_2() {
     let count = Arc::new(AtomicUsize::new(0));
     let count2 = count.clone();
-    let (tx, rx) = ::oneshot::channel();
-    let tx = Arc::new(Mutex::new(Some(tx)));
 
     let srv = test_server(move || {
-        let tx = tx.clone();
         let count = count2.clone();
         HttpService::build().finish(move |_: Request| {
             let count = count.clone();
             count.fetch_add(1, Ordering::Relaxed);
-            if count.load(Ordering::Relaxed) == 2 {
-                let _ = tx.lock().unwrap().take().unwrap().send(());
-            }
             async move {
                 sleep(Millis(1000)).await;
                 count.fetch_sub(1, Ordering::Relaxed);
@@ -835,14 +790,17 @@ async fn test_h1_gracefull_shutdown_2() {
     let mut stream2 = net::TcpStream::connect(srv.addr()).unwrap();
     let _ = stream2.write_all(b"GET /index.html HTTP/1.1\r\n\r\n");
 
-    let _ = rx.await;
-    assert_eq!(count.load(Ordering::Acquire), 2);
+    sleep(Millis(150)).await;
+    assert_eq!(count.load(Ordering::Relaxed), 2);
 
     let (tx, rx) = oneshot::channel();
     rt::spawn(async move {
         srv.stop().await;
         let _ = tx.send(());
     });
+    sleep(Millis(150)).await;
+    assert_eq!(count.load(Ordering::Relaxed), 2);
+
     let _ = rx.await;
     assert_eq!(count.load(Ordering::Relaxed), 0);
 }

ntex/tests/web_server.rs

@@ -1,5 +1,6 @@
 use std::{future::Future, io, io::Read, io::Write, pin::Pin, task::Context, task::Poll};
 
+use brotli2::write::{BrotliDecoder, BrotliEncoder};
 use flate2::read::GzDecoder;
 use flate2::write::{GzEncoder, ZlibDecoder, ZlibEncoder};
 use flate2::Compression;
@@ -317,6 +318,36 @@ async fn test_body_chunked_implicit() {
     assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
 }
 
+#[ntex::test]
+async fn test_body_br_streaming() {
+    let srv = test::server_with(test::config().h1(), || {
+        App::new().wrap(Compress::new(ContentEncoding::Br)).service(
+            web::resource("/").route(web::to(move || async {
+                HttpResponse::Ok()
+                    .streaming(TestBody::new(Bytes::from_static(STR.as_ref()), 24))
+            })),
+        )
+    });
+
+    let mut response = srv
+        .get("/")
+        .header(ACCEPT_ENCODING, "br")
+        .no_decompress()
+        .send()
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+
+    // decode br
+    let mut e = BrotliDecoder::new(Vec::with_capacity(2048));
+    e.write_all(bytes.as_ref()).unwrap();
+    let dec = e.finish().unwrap();
+    assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
+}
+
 #[ntex::test]
 async fn test_head_binary() {
     let srv = test::server_with(test::config().h1(), || {
@@ -391,6 +422,35 @@ async fn test_body_deflate() {
     assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
 }
 
+#[ntex::test]
+async fn test_body_brotli() {
+    let srv = test::server_with(test::config().h1(), || {
+        App::new().wrap(Compress::new(ContentEncoding::Br)).service(
+            web::resource("/")
+                .route(web::to(move || async { HttpResponse::Ok().body(STR) })),
+        )
+    });
+
+    // client request
+    let mut response = srv
+        .get("/")
+        .header(ACCEPT_ENCODING, "br")
+        .no_decompress()
+        .send()
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+
+    // decode brotli
+    let mut e = BrotliDecoder::new(Vec::with_capacity(2048));
+    e.write_all(bytes.as_ref()).unwrap();
+    let dec = e.finish().unwrap();
+    assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
+}
+
 #[ntex::test]
 async fn test_encoding() {
     let srv = test::server_with(test::config().h1(), || {
@@ -584,6 +644,204 @@ async fn test_reading_deflate_encoding_large_random() {
     assert_eq!(bytes, Bytes::from(data));
 }
 
+#[ntex::test]
+async fn test_brotli_encoding() {
+    let srv = test::server_with(test::config().h1(), || {
+        App::new().service(web::resource("/").route(web::to(move |body: Bytes| async {
+            HttpResponse::Ok().body(body)
+        })))
+    });
+
+    let mut e = BrotliEncoder::new(Vec::new(), 5);
+    e.write_all(STR.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let request = srv
+        .post("/")
+        .header(CONTENT_ENCODING, "br")
+        .send_body(enc.clone());
+    let mut response = request.await.unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
+}
+
+#[ntex::test]
+async fn test_brotli_encoding_large() {
+    let data = rand::thread_rng()
+        .sample_iter(&Alphanumeric)
+        .take(320_000)
+        .map(char::from)
+        .collect::<String>();
+
+    let srv = test::server_with(test::config().h1(), || {
+        App::new().service(
+            web::resource("/")
+                .state(web::types::PayloadConfig::new(320_000))
+                .route(web::to(move |body: Bytes| async {
+                    HttpResponse::Ok().streaming(TestBody::new(body, 10240))
+                })),
+        )
+    });
+
+    let mut e = BrotliEncoder::new(Vec::new(), 5);
+    e.write_all(data.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let request = srv
+        .post("/")
+        .header(CONTENT_ENCODING, "br")
+        .send_body(enc.clone());
+    let mut response = request.await.unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().limit(320_000).await.unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
+#[cfg(feature = "openssl")]
+#[ntex::test]
+async fn test_brotli_encoding_large_openssl() {
+    // load ssl keys
+    use tls_openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
+
+    let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
+    builder
+        .set_private_key_file("./tests/key.pem", SslFiletype::PEM)
+        .unwrap();
+    builder
+        .set_certificate_chain_file("./tests/cert.pem")
+        .unwrap();
+
+    let data = STR.repeat(10);
+    let srv = test::server_with(test::config().openssl(builder.build()), move || {
+        App::new().service(web::resource("/").route(web::to(|bytes: Bytes| async {
+            HttpResponse::Ok()
+                .encoding(ContentEncoding::Identity)
+                .body(bytes)
+        })))
+    });
+
+    // body
+    let mut e = BrotliEncoder::new(Vec::new(), 3);
+    e.write_all(data.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let mut response = srv
+        .post("/")
+        .header(CONTENT_ENCODING, "br")
+        .send_body(enc)
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
+#[cfg(feature = "openssl")]
+#[ntex::test]
+async fn test_brotli_encoding_large_openssl_h1() {
+    // load ssl keys
+    use tls_openssl::ssl::{SslAcceptor, SslFiletype, SslMethod};
+
+    let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
+    builder
+        .set_private_key_file("./tests/key.pem", SslFiletype::PEM)
+        .unwrap();
+    builder
+        .set_certificate_chain_file("./tests/cert.pem")
+        .unwrap();
+
+    let data = STR.repeat(10);
+    let srv = test::server_with(test::config().openssl(builder.build()).h1(), move || {
+        App::new().service(web::resource("/").route(web::to(|bytes: Bytes| async {
+            HttpResponse::Ok()
+                .encoding(ContentEncoding::Identity)
+                .body(bytes)
+        })))
+    });
+
+    // body
+    let mut e = BrotliEncoder::new(Vec::new(), 3);
+    e.write_all(data.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let mut response = srv
+        .post("/")
+        .header(CONTENT_ENCODING, "br")
+        .send_body(enc)
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
+#[cfg(feature = "openssl")]
+#[ntex::test]
+async fn test_brotli_encoding_large_openssl_h2() {
+    // load ssl keys
+    use tls_openssl::ssl::{AlpnError, SslAcceptor, SslFiletype, SslMethod};
+
+    let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
+    builder
+        .set_private_key_file("./tests/key.pem", SslFiletype::PEM)
+        .unwrap();
+    builder
+        .set_certificate_chain_file("./tests/cert.pem")
+        .unwrap();
+    builder.set_alpn_select_callback(|_, protos| {
+        const H2: &[u8] = b"\x02h2";
+        const H11: &[u8] = b"\x08http/1.1";
+        if protos.windows(3).any(|window| window == H2) {
+            Ok(b"h2")
+        } else if protos.windows(9).any(|window| window == H11) {
+            Ok(b"http/1.1")
+        } else {
+            Err(AlpnError::NOACK)
+        }
+    });
+    builder.set_alpn_protos(b"\x08http/1.1\x02h2").unwrap();
+
+    let data = STR.repeat(10);
+    let srv = test::server_with(test::config().openssl(builder.build()).h2(), move || {
+        App::new().service(web::resource("/").route(web::to(|bytes: Bytes| async {
+            HttpResponse::Ok()
+                .encoding(ContentEncoding::Identity)
+                .body(bytes)
+        })))
+    });
+
+    // body
+    let mut e = BrotliEncoder::new(Vec::new(), 3);
+    e.write_all(data.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let mut response = srv
+        .post("/")
+        .header(CONTENT_ENCODING, "br")
+        .send_body(enc)
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
 #[cfg(all(feature = "rustls", feature = "openssl"))]
 #[ntex::test]
 async fn test_reading_deflate_encoding_large_random_rustls() {