PipelineCall is static

CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [0.11.1] - 2023-06-23
+
+* `PipelineCall` is static
+
 ## [0.11.0] - 2023-06-22
 
 * Release v0.11.0

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-mqtt"
-version = "0.11.0"
+version = "0.11.1"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Client and Server framework for MQTT v5 and v3.1.1 protocols"
 documentation = "https://docs.rs/ntex-mqtt"
@@ -12,7 +12,7 @@ exclude = [".gitignore", ".travis.yml", ".cargo/config"]
 edition = "2018"
 
 [dependencies]
-ntex = "0.7.0"
+ntex = "0.7.1"
 bitflags = "1.3"
 log = "0.4"
 pin-project-lite = "0.2"
@@ -26,7 +26,7 @@ ntex-tls = "0.3.0"
 rustls = "0.21"
 rustls-pemfile = "1.0"
 openssl = "0.10"
-ntex = { version = "0.7.0", features = ["tokio", "rustls", "openssl"] }
+ntex = { version = "0.7.1", features = ["tokio", "rustls", "openssl"] }
 test-case = "3"
 
 [profile.dev]

examples/mqtt-ws-server.rs

@@ -124,7 +124,7 @@ async fn main() -> std::io::Result<()> {
                             // and then switch to websokets streaming
                             .upgrade(
                                 // validate ws request and init ws transport
-                                pipeline_factory(
+                                chain_factory(
                                     |(req, io, codec): (Request, Io<_>, h1::Codec)| {
                                         async move {
                                             match ws::handshake(req.head()) {

src/io.rs

@@ -25,7 +25,7 @@ pin_project_lite::pin_project! {
         inner: DispatcherInner<S, U>,
         pool: Pool,
         #[pin]
-        response: Option<PipelineCall<'static, S, DispatchItem<U>>>,
+        response: Option<PipelineCall<S, DispatchItem<U>>>,
         response_idx: usize,
     }
 }
@@ -321,7 +321,7 @@ where
                     if let Some(item) = item {
                         // optimize first call
                         if this.response.is_none() {
-                            this.response.set(Some(this.service.call(item).into_static()));
+                            this.response.set(Some(this.service.call(item)));
                             let res = this.response.as_mut().as_pin_mut().unwrap().poll(cx);
 
                             let mut state = inner.state.borrow_mut();
@@ -360,7 +360,7 @@ where
                             let st = inner.io.get_ref();
                             let codec = this.codec.clone();
                             let state = inner.state.clone();
-                            let fut = this.service.call(item).into_static();
+                            let fut = this.service.call(item);
                             ntex::rt::spawn(async move {
                                 let item = fut.await;
                                 state.borrow_mut().handle_result(

src/v3/client/dispatcher.rs

@@ -90,7 +90,8 @@ where
             self.inner.sink.close();
             let inner = self.inner.clone();
             *shutdown = Some(Box::pin(async move {
-                let _ = Pipeline::new(&inner.control).call(ControlMessage::closed()).await;
+                let _ =
+                    Pipeline::new(&inner.control).service_call(ControlMessage::closed()).await;
             }));
         }
 

src/v3/dispatcher.rs

@@ -144,7 +144,8 @@ where
             self.inner.sink.close();
             let inner = self.inner.clone();
             *shutdown = Some(Box::pin(async move {
-                let _ = Pipeline::new(&inner.control).call(ControlMessage::closed()).await;
+                let _ =
+                    Pipeline::new(&inner.control).service_call(ControlMessage::closed()).await;
             }));
         }
 

src/v5/client/dispatcher.rs

@@ -116,7 +116,8 @@ where
             self.inner.sink.drop_sink();
             let inner = self.inner.clone();
             *shutdown = Some(Box::pin(async move {
-                let _ = Pipeline::new(&inner.control).call(ControlMessage::closed()).await;
+                let _ =
+                    Pipeline::new(&inner.control).service_call(ControlMessage::closed()).await;
             }));
         }
 

src/v5/dispatcher.rs

@@ -149,7 +149,8 @@ where
             self.inner.sink.drop_sink();
             let inner = self.inner.clone();
             *shutdown = Some(Box::pin(async move {
-                let _ = Pipeline::new(&inner.control).call(ControlMessage::closed()).await;
+                let _ =
+                    Pipeline::new(&inner.control).service_call(ControlMessage::closed()).await;
             }));
         }