Use ntex_dispatcher::Dispatcher instead of ntex-io (#78)

Author: fafhrd91

.github/workflows/CI.yml

@@ -105,7 +105,7 @@ jobs:
     strategy:
       matrix:
         rust:
-          - 1.86 # never go past ntex's msrv
+          - 1.88 # never go past ntex's msrv
 
         os:
           - ubuntu-latest

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.5.0] - 2026-01-29
+
+* Use ntex_dispatcher::Dispatcher instead of ntex-io
+
 ## [3.4.0] - 2026-01-17
 
 * Update bytes and codec apis

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-h2"
-version = "3.4.0"
+version = "3.5.0"
 license = "MIT OR Apache-2.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "An HTTP/2 client and server"
@@ -11,7 +11,7 @@ keywords = ["http", "async", "non-blocking"]
 categories = ["asynchronous", "web-programming", "network-programming"]
 exclude = ["fixtures/**", "util/**", "fuzz/**"]
 edition = "2024"
-rust-version = "1.86"
+rust-version = "1.88"
 
 [features]
 default = []
@@ -23,10 +23,11 @@ unstable = []
 features = []
 
 [dependencies]
-ntex-bytes = "1.3"
+ntex-bytes = "1.4"
 ntex-codec = "1.1"
+ntex-dispatcher = "3"
 ntex-http = "1"
-ntex-io = "3.4"
+ntex-io = "3.5"
 ntex-net = "3"
 ntex-service = "4"
 ntex-util = "3"
@@ -49,7 +50,7 @@ walkdir = "2.3.2"
 serde = "1"
 serde_json = "1"
 
-ntex = { version = "3.0.0-pre.11", features = ["openssl"] }
+ntex = { version = "3.0.0-pre.14", features = ["openssl"] }
 ntex-tls = { version = "3", features = ["openssl"] }
 openssl = "0.10"
 

README.md

@@ -5,7 +5,7 @@ HTTP/2 client & server implementation for Rust.
 [![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)
 [![Crates.io](https://img.shields.io/crates/v/ntex-h2.svg)](https://crates.io/crates/ntex-h2)
 [![Documentation](https://img.shields.io/docsrs/ntex-h2/latest)](https://docs.rs/ntex-h2) 
-[![Version](https://img.shields.io/badge/rustc-1.86+-lightgray.svg)](https://releases.rs/docs/1.85.0/) 
+[![Version](https://img.shields.io/badge/rustc-1.88+-lightgray.svg)](https://releases.rs/docs/1.88.0/) 
 [![CI](https://github.com/ntex-rs/ntex-h2/actions/workflows/CI.yml/badge.svg)](https://github.com/ntex-rs/ntex-h2/actions/workflows/CI.yml) 
 [![codecov](https://codecov.io/gh/ntex-rs/ntex-h2/branch/main/graph/badge.svg)](https://codecov.io/gh/ntex-rs/ntex-h2) 
 

src/client/simple.rs

@@ -2,8 +2,9 @@ use std::{fmt, future::Future, pin::Pin, rc::Rc, task::Context, task::Poll};
 
 use nanorand::Rng;
 use ntex_bytes::ByteString;
+use ntex_dispatcher::Dispatcher as IoDispatcher;
 use ntex_http::{HeaderMap, Method, uri::Scheme};
-use ntex_io::{Dispatcher as IoDispatcher, IoBoxed, IoRef, OnDisconnect};
+use ntex_io::{IoBoxed, IoRef, OnDisconnect};
 use ntex_service::cfg::Cfg;
 use ntex_util::{channel::pool, time::Millis, time::Sleep};
 
@@ -266,11 +267,11 @@ impl Future for ClientDisconnect {
 
         if Pin::new(&mut this.disconnect).poll(cx).is_ready() {
             return Poll::Ready(this.client.0.con.check_error_with_disconnect());
-        } else if let Some(ref mut sleep) = this.timeout {
-            if sleep.poll_elapsed(cx).is_ready() {
-                this.client.0.con.close();
-                return Poll::Ready(Err(OperationError::Disconnected));
-            }
+        } else if let Some(ref mut sleep) = this.timeout
+            && sleep.poll_elapsed(cx).is_ready()
+        {
+            this.client.0.con.close();
+            return Poll::Ready(Err(OperationError::Disconnected));
         }
         Poll::Pending
     }

src/connection.rs

@@ -403,13 +403,13 @@ impl Connection {
                 } else {
                     let local = self.0.active_local_streams.get();
                     self.0.active_local_streams.set(local - 1);
-                    if let Some(max) = self.0.local_max_concurrent_streams.get() {
-                        if local == max {
-                            while let Some(tx) = self.0.readiness.borrow_mut().pop_front() {
-                                if !tx.is_canceled() {
-                                    let _ = tx.send(());
-                                    break;
-                                }
+                    if let Some(max) = self.0.local_max_concurrent_streams.get()
+                        && local == max
+                    {
+                        while let Some(tx) = self.0.readiness.borrow_mut().pop_front() {
+                            if !tx.is_canceled() {
+                                let _ = tx.send(());
+                                break;
                             }
                         }
                     }
@@ -539,18 +539,18 @@ impl RecvHalfConnection {
             }
 
             // Max concurrent streams check
-            if let Some(max) = self.0.local_config.remote_max_concurrent_streams {
-                if self.0.active_remote_streams.get() >= max {
-                    // check if client opened more streams than allowed
-                    // in that case close connection
-                    return if self.flags().contains(ConnectionFlags::STREAM_REFUSED) {
-                        Err(Either::Left(ConnectionError::ConcurrencyOverflow))
-                    } else {
-                        self.encode(frame::Reset::new(id, frame::Reason::REFUSED_STREAM));
-                        self.set_flags(ConnectionFlags::STREAM_REFUSED);
-                        Ok(None)
-                    };
-                }
+            if let Some(max) = self.0.local_config.remote_max_concurrent_streams
+                && self.0.active_remote_streams.get() >= max
+            {
+                // check if client opened more streams than allowed
+                // in that case close connection
+                return if self.flags().contains(ConnectionFlags::STREAM_REFUSED) {
+                    Err(Either::Left(ConnectionError::ConcurrencyOverflow))
+                } else {
+                    self.encode(frame::Reset::new(id, frame::Reason::REFUSED_STREAM));
+                    self.set_flags(ConnectionFlags::STREAM_REFUSED);
+                    Ok(None)
+                };
             }
 
             // Pseudo-headers validation

src/dispatcher.rs

@@ -1,9 +1,8 @@
 use std::{fmt, future::Future, future::poll_fn, rc::Rc, task::Context, task::Poll};
 
-use ntex_io::DispatchItem;
+use ntex_dispatcher::{DispatchItem, Reason as DispReason};
 use ntex_service::{Pipeline, Service, ServiceCtx};
-use ntex_util::future::{Either, join};
-use ntex_util::{HashMap, spawn};
+use ntex_util::{HashMap, future::Either, future::join, spawn};
 
 use crate::connection::{Connection, RecvHalfConnection};
 use crate::control::{Control, ControlAck};
@@ -247,19 +246,19 @@ where
                     Ok(None)
                 }
             },
-            DispatchItem::EncoderError(err) => {
+            DispatchItem::Stop(DispReason::Encoder(err)) => {
                 let err = ConnectionError::from(err);
                 let streams = self.connection.proto_error(&err);
                 self.handle_connection_error(streams, err.into());
                 control(Control::proto_error(err), &self.inner, ctx).await
             }
-            DispatchItem::DecoderError(err) => {
+            DispatchItem::Stop(DispReason::Decoder(err)) => {
                 let err = ConnectionError::from(err);
                 let streams = self.connection.proto_error(&err);
                 self.handle_connection_error(streams, err.into());
                 control(Control::proto_error(err), &self.inner, ctx).await
             }
-            DispatchItem::KeepAliveTimeout => {
+            DispatchItem::Stop(DispReason::KeepAliveTimeout) => {
                 log::warn!(
                     "{}: did not receive pong response in time, closing connection",
                     self.connection.tag(),
@@ -273,7 +272,7 @@ where
                 )
                 .await
             }
-            DispatchItem::ReadTimeout => {
+            DispatchItem::Stop(DispReason::ReadTimeout) => {
                 log::warn!(
                     "{}: did not receive complete frame in time, closing connection",
                     self.connection.tag(),
@@ -287,12 +286,12 @@ where
                 )
                 .await
             }
-            DispatchItem::Disconnect(err) => {
+            DispatchItem::Stop(DispReason::Io(err)) => {
                 let streams = self.connection.disconnect();
                 self.handle_connection_error(streams, OperationError::Disconnected);
                 control(Control::peer_gone(err), &self.inner, ctx).await
             }
-            DispatchItem::WBackPressureEnabled | DispatchItem::WBackPressureDisabled => Ok(None),
+            DispatchItem::Control(_) => Ok(None),
         }
     }
 }
@@ -346,10 +345,10 @@ where
 {
     match ctx.call(inner.control.get_ref(), pkt).await {
         Ok(res) => {
-            if let Some(Frame::Reset(ref rst)) = res.frame {
-                if !rst.stream_id().is_zero() {
-                    inner.connection.rst_stream(rst.stream_id(), rst.reason());
-                }
+            if let Some(Frame::Reset(ref rst)) = res.frame
+                && !rst.stream_id().is_zero()
+            {
+                inner.connection.rst_stream(rst.stream_id(), rst.reason());
             }
             if let Some(frm) = res.frame {
                 inner.connection.encode(frm);

src/frame/settings.rs

@@ -148,7 +148,7 @@ impl Settings {
         }
 
         // Ensure the payload length is correct, each setting is 6 bytes long.
-        if payload.len() % 6 != 0 {
+        if !payload.len().is_multiple_of(6) {
             log::debug!("invalid settings payload length; len={:?}", payload.len());
             return Err(FrameError::InvalidPayloadAckSettings);
         }

src/frame/stream_id.rs

@@ -43,14 +43,14 @@ impl StreamId {
     /// was initiated by the client.
     pub const fn is_client_initiated(&self) -> bool {
         let id = self.0;
-        id != 0 && id % 2 == 1
+        id != 0 && !id.is_multiple_of(2)
     }
 
     /// Returns true if this stream ID corresponds to a stream that
     /// was initiated by the server.
     pub const fn is_server_initiated(&self) -> bool {
         let id = self.0;
-        id != 0 && id % 2 == 0
+        id != 0 && id.is_multiple_of(2)
     }
 
     /// Return a new `StreamId` for stream 0.

src/hpack/table.rs

@@ -469,11 +469,11 @@ impl Table {
         let mut first_ideal = 0;
 
         for (i, pos) in self.indices.iter().enumerate() {
-            if let Some(pos) = *pos {
-                if 0 == probe_distance(self.mask, pos.hash, i) {
-                    first_ideal = i;
-                    break;
-                }
+            if let Some(pos) = *pos
+                && 0 == probe_distance(self.mask, pos.hash, i)
+            {
+                first_ideal = i;
+                break;
             }
         }