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

Author: fafhrd91

.github/workflows/linux.yml

@@ -8,7 +8,7 @@ jobs:
       fail-fast: false
       matrix:
         version:
-          - 1.86.0 # MSRV
+          - 1.88.0 # MSRV
           - stable
           - nightly
 

CHANGES.md

@@ -1,6 +1,10 @@
 # Changes
 
-## [5.3.0] - 2026-01-xx
+## [5.4.0] - 2026-01-29
+
+* Use ntex_dispatcher::Dispatcher instead of ntex-io
+
+## [5.3.0] - 2026-01-26
 
 * Add `Begin` frame to `Session`
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-amqp"
-version = "5.3.0"
+version = "5.4.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "AMQP 1.0 Client/Server framework"
 documentation = "https://docs.rs/ntex-amqp"
@@ -10,7 +10,7 @@ keywords = ["AMQP", "IoT", "messaging"]
 license = "MIT OR Apache-2.0"
 exclude = [".gitignore", ".travis.yml", ".cargo/config"]
 edition = "2024"
-rust-version = "1.86"
+rust-version = "1.88"
 
 [workspace]
 members = [
@@ -28,9 +28,10 @@ frame-trace = []
 ntex-amqp-codec = "2.0.0"
 
 ntex-bytes = "1.4"
+ntex-dispatcher = "3"
 ntex-router = "1"
-ntex-rt = "3.4"
-ntex-io = "3.3"
+ntex-rt = "3.5"
+ntex-io = "3.5"
 ntex-net = "3.5"
 ntex-util = "3.4"
 ntex-service = "4"
@@ -47,7 +48,7 @@ thiserror = "2"
 env_logger = "0.11"
 rand = "0.9"
 ntex-amqp = { path = ".", features = ["frame-trace"] }
-ntex = "3.0.0-pre.11"
+ntex = "3.0.0-pre.14"
 
 [patch.crates-io]
 ntex-amqp = { path = "." }

codec/Cargo.toml

@@ -1,14 +1,14 @@
 [package]
 name = "ntex-amqp-codec"
-version = "2.1.0"
+version = "2.1.1"
 description = "AMQP 1.0 Protocol Codec"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>", "Max Gortman <mgortman@microsoft.com>", "Mike Yagley <myagley@gmail.com>"]
 license = "MIT/Apache-2.0"
 edition = "2024"
-rust-version = "1.86"
+rust-version = "1.88"
 
 [dependencies]
-ntex-bytes = "1.3"
+ntex-bytes = "1.4"
 ntex-codec = "1.1"
 ntex-util = "3"
 byteorder = "1"

examples/server.rs

@@ -13,7 +13,7 @@ async fn main() -> std::io::Result<()> {
     // std::env::set_var("RUST_LOG", "trace,ntex_io=info");
     env_logger::init();
 
-    ntex::server::Server::build()
+    ntex::server::Server::builder()
         .bind("amqp", "127.0.0.1:5671", async |_| {
             server::Server::build(|con: server::Handshake| async move {
                 match con {

src/client/connection.rs

@@ -1,4 +1,5 @@
-use ntex_io::{Dispatcher as IoDispatcher, IoBoxed};
+use ntex_dispatcher::Dispatcher as IoDispatcher;
+use ntex_io::IoBoxed;
 use ntex_service::{IntoService, Pipeline, Service, fn_service};
 use ntex_util::future::Ready;
 

src/connection.rs

@@ -182,10 +182,10 @@ impl ConnectionRef {
     }
 
     pub(crate) fn close_session(&self, id: usize) {
-        if let Some(state) = self.0.get_mut().sessions.get_mut(id) {
-            if let SessionState::Established(inner) = state {
-                *state = SessionState::Closing(inner.clone());
-            }
+        if let Some(state) = self.0.get_mut().sessions.get_mut(id)
+            && let SessionState::Established(inner) = state
+        {
+            *state = SessionState::Closing(inner.clone());
         }
     }
 

src/delivery.rs

@@ -330,10 +330,10 @@ impl TransferBuilder {
         } else if inner.closed {
             Err(AmqpProtocolError::Disconnected)
         } else {
-            if let Some(limit) = inner.max_message_size {
-                if self.data.len() > limit as usize {
-                    return Err(AmqpProtocolError::BodyTooLarge);
-                }
+            if let Some(limit) = inner.max_message_size
+                && self.data.len() > limit as usize
+            {
+                return Err(AmqpProtocolError::BodyTooLarge);
             }
 
             let (id, tag) = self

src/dispatcher.rs

@@ -1,7 +1,7 @@
 use std::task::{Context, Poll, ready};
 use std::{cell, cmp, future::Future, future::poll_fn, marker, pin::Pin};
 
-use ntex_io::DispatchItem;
+use ntex_dispatcher::{DispatchItem, Reason};
 use ntex_rt::spawn;
 use ntex_service::{Pipeline, PipelineBinding, PipelineCall, Service, ServiceCtx};
 use ntex_util::time::{Millis, Sleep, sleep};
@@ -261,31 +261,31 @@ where
 
                 Ok(None)
             }
-            DispatchItem::EncoderError(err) | DispatchItem::DecoderError(err) => {
+            DispatchItem::Stop(Reason::Encoder(err)) | DispatchItem::Stop(Reason::Decoder(err)) => {
                 self.call_control_service(ControlFrame::new_kind(ControlFrameKind::ProtocolError(
                     err.into(),
                 )));
                 Ok(None)
             }
-            DispatchItem::KeepAliveTimeout => {
+            DispatchItem::Stop(Reason::KeepAliveTimeout) => {
                 self.call_control_service(ControlFrame::new_kind(ControlFrameKind::ProtocolError(
                     AmqpProtocolError::KeepAliveTimeout,
                 )));
                 Ok(None)
             }
-            DispatchItem::ReadTimeout => {
+            DispatchItem::Stop(Reason::ReadTimeout) => {
                 self.call_control_service(ControlFrame::new_kind(ControlFrameKind::ProtocolError(
                     AmqpProtocolError::ReadTimeout,
                 )));
                 Ok(None)
             }
-            DispatchItem::Disconnect(e) => {
+            DispatchItem::Stop(Reason::Io(e)) => {
                 self.call_control_service(ControlFrame::new_kind(ControlFrameKind::Disconnected(
                     e,
                 )));
                 Ok(None)
             }
-            DispatchItem::WBackPressureEnabled | DispatchItem::WBackPressureDisabled => Ok(None),
+            DispatchItem::Control(_) => Ok(None),
         }
     }
 }

src/router.rs

@@ -182,10 +182,10 @@ impl<S: 'static> Service<Message> for RouterService<S> {
                 Ok(())
             }
             Message::Transfer(link) => {
-                if let Some(Some(_)) = self.0.get_ref().handlers.get(&link) {
-                    if let Some((delivery, tr)) = link.get_delivery() {
-                        service_call(link, delivery, tr, &self.0).await?;
-                    }
+                if let Some(Some(_)) = self.0.get_ref().handlers.get(&link)
+                    && let Some((delivery, tr)) = link.get_delivery()
+                {
+                    service_call(link, delivery, tr, &self.0).await?;
                 }
                 Ok(())
             }