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

Author: fafhrd91

Cargo.toml

@@ -42,10 +42,10 @@ ntex-macros = { path = "ntex-macros" }
 ntex-util = { path = "ntex-util" }
 
 [workspace.dependencies]
-ntex = "3.0.0-pre.14"
+ntex = "3.0.0-pre.15"
 ntex-bytes = "1.4.1"
 ntex-codec = "1.1.0"
-ntex-io = "3.5.0"
+ntex-io = "3.6.0"
 ntex-dispatcher = "3.0.0"
 ntex-net = "3.5.1"
 ntex-http = "1.0.0"
@@ -56,7 +56,7 @@ ntex-service = "4.0.0"
 ntex-tls = "3.2.0"
 ntex-macros = "3.1.0"
 ntex-util = "3.4.0"
-ntex-h2 = "3.4.0"
+ntex-h2 = "3.5.0"
 
 ntex-polling = "3.10.0"
 ntex-io-uring = "0.7.120"

ntex-io/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.6.0] - 2026-01-29
+
+* Deprecated Dispatcher
+
 ## [3.5.0] - 2026-01-21
 
 * Set IoConfig ref lifetime

ntex-io/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-io"
-version = "3.5.0"
+version = "3.6.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Utilities for abstracting io streams"
 keywords = ["network", "framework", "async", "futures"]

ntex-io/src/dispatcher.rs

@@ -1,5 +1,5 @@
 //! Framed transport dispatcher
-#![allow(clippy::let_underscore_future)]
+#![allow(deprecated, clippy::let_underscore_future)]
 use std::task::{Context, Poll, ready};
 use std::{cell::Cell, future::Future, pin::Pin, rc::Rc};
 
@@ -14,6 +14,7 @@ type Response<U> = <U as Encoder>::Item;
 pin_project_lite::pin_project! {
     /// Dispatcher - is a future that reads frames from bytes stream
     /// and pass then to the service.
+    #[deprecated]
     pub struct Dispatcher<S, U>
     where
         S: Service<DispatchItem<U>, Response = Option<Response<U>>>,

ntex-io/src/lib.rs

@@ -1,6 +1,6 @@
 //! Utilities for abstructing io streams
 #![deny(rust_2018_idioms, unreachable_pub, missing_debug_implementations)]
-#![allow(async_fn_in_trait)]
+#![allow(deprecated, async_fn_in_trait)]
 
 use std::io::{Error as IoError, Result as IoResult};
 use std::{any::Any, any::TypeId, fmt, task::Context, task::Poll};
@@ -176,6 +176,7 @@ where
     }
 }
 
+#[deprecated]
 /// Dispatcher item
 pub enum DispatchItem<U: Encoder + Decoder> {
     Item(<U as Decoder>::Item),

ntex/CHANGES.md

@@ -1,6 +1,10 @@
 # Changes
 
-## [3.0.0-pre.14] 2026-01-xx
+## [3.0.0-pre.15] 2026-01-29
+
+* Use ntex_dispatcher::Dispatcher instead of ntex-io
+
+## [3.0.0-pre.14] 2026-01-28
 
 * MSRV 1.88
 

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "3.0.0-pre.14"
+version = "3.0.0-pre.15"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"
@@ -72,6 +72,7 @@ no-test-logging = []
 [dependencies]
 ntex-bytes = { workspace = true }
 ntex-codec = { workspace = true }
+ntex-dispatcher = { workspace = true }
 ntex-http = { workspace = true }
 ntex-router = { workspace = true }
 ntex-service = { workspace = true }

ntex/src/lib.rs

@@ -104,7 +104,14 @@ pub mod time {
 
 pub mod io {
     //! IO streaming utilities.
-    pub use ntex_io::*;
+    pub use ntex_dispatcher::*;
+
+    pub use ntex_io::{
+        Base, Decoded, Filter, FilterCtx, FilterLayer, FilterReadStatus, Flags, Framed,
+        Handle, Io, IoBoxed, IoConfig, IoContext, IoRef, IoStatusUpdate, IoStream,
+        IoTaskStatus, Layer, OnDisconnect, ReadBuf, Readiness, RecvError, Sealed,
+        TimerHandle, WriteBuf, cfg, seal, testing, types,
+    };
 }
 
 pub mod testing {

ntex/src/web/ws.rs

@@ -4,14 +4,13 @@ use std::{fmt, rc::Rc};
 pub use crate::ws::{CloseCode, CloseReason, Frame, Message, WsSink};
 
 use crate::http::{StatusCode, body::BodySize, h1};
+use crate::io::{DispatchItem, IoConfig, Reason};
 use crate::service::{
-    IntoServiceFactory, ServiceFactory, apply_fn, chain_factory, fn_factory_with_config,
+    IntoServiceFactory, ServiceFactory, chain_factory, fn_factory_with_config, fn_service,
 };
 use crate::web::{HttpRequest, HttpResponse};
 use crate::ws::{self, error::HandshakeError, error::WsError, handshake};
-use crate::{
-    SharedCfg, io::DispatchItem, io::IoConfig, rt, time::Seconds, util::Either, util::Ready,
-};
+use crate::{SharedCfg, rt, time::Seconds};
 
 thread_local! {
     static CFG: SharedCfg = SharedCfg::new("WS")
@@ -29,44 +28,30 @@ where
 {
     let inner_factory = Rc::new(chain_factory(factory).map_err(WsError::Service));
 
-    let factory = fn_factory_with_config(move |sink: WsSink| {
-        let factory = inner_factory.clone();
-
-        async move {
-            let srv = factory.create(sink.clone()).await?;
-            let sink = sink.clone();
-
-            Ok::<_, T::InitError>(apply_fn(srv, move |req, srv| match req {
-                DispatchItem::<ws::Codec>::Item(item) => {
-                    let s = if matches!(item, Frame::Close(_)) {
-                        Some(sink.clone())
-                    } else {
-                        None
-                    };
-                    Either::Left(async move {
-                        let result = srv.call(item).await;
-                        if let Some(s) = s {
-                            let _ = rt::spawn(async move { s.io().close() });
-                        }
-                        result
-                    })
-                }
-                DispatchItem::WBackPressureEnabled
-                | DispatchItem::WBackPressureDisabled => Either::Right(Ready::Ok(None)),
-                DispatchItem::KeepAliveTimeout => {
-                    Either::Right(Ready::Err(WsError::KeepAlive))
-                }
-                DispatchItem::ReadTimeout => {
-                    Either::Right(Ready::Err(WsError::ReadTimeout))
-                }
-                DispatchItem::DecoderError(e) | DispatchItem::EncoderError(e) => {
-                    Either::Right(Ready::Err(WsError::Protocol(e)))
-                }
-                DispatchItem::Disconnect(e) => {
-                    Either::Right(Ready::Err(WsError::Disconnected(e)))
+    let factory = fn_factory_with_config(async move |sink: WsSink| {
+        let srv = inner_factory.pipeline(sink.clone()).await?;
+        let sink = sink.clone();
+
+        Ok::<_, T::InitError>(fn_service(async move |req| match req {
+            DispatchItem::<ws::Codec>::Item(item) => {
+                let s = if matches!(item, Frame::Close(_)) {
+                    Some(sink.clone())
+                } else {
+                    None
+                };
+                let result = srv.call(item).await;
+                if let Some(s) = s {
+                    let _ = rt::spawn(async move { s.io().close() });
                 }
-            }))
-        }
+                result
+            }
+            DispatchItem::Control(_) => Ok(None),
+            DispatchItem::Stop(Reason::KeepAliveTimeout) => Err(WsError::KeepAlive),
+            DispatchItem::Stop(Reason::ReadTimeout) => Err(WsError::ReadTimeout),
+            DispatchItem::Stop(Reason::Decoder(e))
+            | DispatchItem::Stop(Reason::Encoder(e)) => Err(WsError::Protocol(e)),
+            DispatchItem::Stop(Reason::Io(e)) => Err(WsError::Disconnected(e)),
+        }))
     });
 
     start_with(req, factory).await

ntex/src/ws/client.rs

@@ -16,7 +16,7 @@ use crate::connect::{Connect, ConnectError, Connector};
 use crate::http::header::{self, AUTHORIZATION, HeaderMap, HeaderName, HeaderValue};
 use crate::http::{ConnectionType, RequestHead, RequestHeadType, StatusCode, Uri};
 use crate::http::{body::BodySize, error::HttpError, h1};
-use crate::io::{Base, DispatchItem, Dispatcher, Filter, Io, Layer, Sealed};
+use crate::io::{Base, DispatchItem, Dispatcher, Filter, Io, Layer, Reason, Sealed};
 use crate::service::{IntoService, Pipeline, apply_fn, fn_service};
 use crate::time::{Millis, timeout};
 use crate::{Service, ServiceFactory, SharedCfg, channel::mpsc, rt, util::Ready, ws};
@@ -743,14 +743,12 @@ impl WsConnection<Sealed> {
             |req, svc| async move {
                 match req {
                     DispatchItem::<ws::Codec>::Item(item) => svc.call(item).await,
-                    DispatchItem::WBackPressureEnabled
-                    | DispatchItem::WBackPressureDisabled => Ok(None),
-                    DispatchItem::KeepAliveTimeout => Err(WsError::KeepAlive),
-                    DispatchItem::ReadTimeout => Err(WsError::ReadTimeout),
-                    DispatchItem::DecoderError(e) | DispatchItem::EncoderError(e) => {
-                        Err(WsError::Protocol(e))
-                    }
-                    DispatchItem::Disconnect(e) => Err(WsError::Disconnected(e)),
+                    DispatchItem::Control(_) => Ok(None),
+                    DispatchItem::Stop(Reason::KeepAliveTimeout) => Err(WsError::KeepAlive),
+                    DispatchItem::Stop(Reason::ReadTimeout) => Err(WsError::ReadTimeout),
+                    DispatchItem::Stop(Reason::Decoder(e))
+                    | DispatchItem::Stop(Reason::Encoder(e)) => Err(WsError::Protocol(e)),
+                    DispatchItem::Stop(Reason::Io(e)) => Err(WsError::Disconnected(e)),
                 }
             },
         );