Add Service impl for Pipeline, PipelineSvc (#694)

Author: fafhrd91

.github/workflows/windows.yml

@@ -60,8 +60,9 @@ jobs:
 
       - name: Install OpenSSL
         run: |
+          $env:VCPKG_MANIFEST_MODE = 'OFF'
           vcpkg integrate install
-          vcpkg install openssl:x64-windows
+          vcpkg install --clean-after-build openssl:x64-windows
           Copy-Item C:\vcpkg\installed\x64-windows\bin\libcrypto-3-x64.dll C:\vcpkg\installed\x64-windows\bin\libcrypto.dll
           Copy-Item C:\vcpkg\installed\x64-windows\bin\libssl-3-x64.dll C:\vcpkg\installed\x64-windows\bin\libssl.dll
           Get-ChildItem C:\vcpkg\installed\x64-windows\bin

ntex-service/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [3.8.0] - 2025-12-16
+
+* Add Service impl for Pipeline, PipelineSvc
+
 ## [3.7.3] - 2025-12-14
 
 * Fix handling nested readiness checks

ntex-service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-service"
-version = "3.7.3"
+version = "3.8.0"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "ntex service"
 keywords = ["network", "framework", "async", "futures"]
@@ -25,5 +25,5 @@ slab = { workspace = true }
 foldhash = { workspace = true }
 
 [dev-dependencies]
-ntex = "3.0.0-pre.6"
+ntex = "3.0.0-pre.7"
 ntex-util = "3"

ntex-service/src/lib.rs

@@ -33,7 +33,7 @@ pub use self::fn_service::{fn_factory, fn_factory_with_config, fn_service};
 pub use self::fn_shutdown::fn_shutdown;
 pub use self::map_config::{map_config, unit_config};
 pub use self::middleware::{Identity, Middleware, Middleware2, Stack, apply, apply2};
-pub use self::pipeline::{Pipeline, PipelineBinding, PipelineCall};
+pub use self::pipeline::{Pipeline, PipelineBinding, PipelineCall, PipelineSvc};
 
 #[allow(unused_variables)]
 /// An asynchronous function of `Request` to a `Response`.

ntex-service/src/pipeline.rs

@@ -1,6 +1,6 @@
 use std::{cell, fmt, future::Future, marker, pin::Pin, rc::Rc, task::Context, task::Poll};
 
-use crate::{Service, ServiceCtx, ctx::WaitersRef};
+use crate::{IntoService, Service, ServiceCtx, ctx::WaitersRef};
 
 #[derive(Debug)]
 /// Container for a service.
@@ -167,6 +167,79 @@ impl<S: fmt::Debug> fmt::Debug for PipelineState<S> {
     }
 }
 
+#[derive(Debug)]
+/// Service wrapper for Pipeline
+pub struct PipelineSvc<S> {
+    inner: Pipeline<S>,
+}
+
+impl<S> PipelineSvc<S> {
+    #[inline]
+    /// Construct new PipelineSvc
+    pub fn new(inner: Pipeline<S>) -> Self {
+        Self { inner }
+    }
+}
+
+impl<S, Req> Service<Req> for PipelineSvc<S>
+where
+    S: Service<Req>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+
+    #[inline]
+    async fn call(
+        &self,
+        req: Req,
+        _: ServiceCtx<'_, Self>,
+    ) -> Result<Self::Response, Self::Error> {
+        self.inner.call(req).await
+    }
+
+    #[inline]
+    async fn ready(&self, _: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
+        self.inner.ready().await
+    }
+
+    #[inline]
+    async fn shutdown(&self) {
+        self.inner.shutdown().await
+    }
+
+    #[inline]
+    fn poll(&self, cx: &mut Context<'_>) -> Result<(), Self::Error> {
+        self.inner.poll(cx)
+    }
+}
+
+impl<S> From<S> for PipelineSvc<S> {
+    #[inline]
+    fn from(svc: S) -> Self {
+        PipelineSvc {
+            inner: Pipeline::new(svc),
+        }
+    }
+}
+
+impl<S> Clone for PipelineSvc<S> {
+    fn clone(&self) -> Self {
+        PipelineSvc {
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<S, R> IntoService<PipelineSvc<S>, R> for Pipeline<S>
+where
+    S: Service<R>,
+{
+    #[inline]
+    fn into_service(self) -> PipelineSvc<S> {
+        PipelineSvc::new(self)
+    }
+}
+
 /// Bound container for a service.
 pub struct PipelineBinding<S, R>
 where
@@ -421,3 +494,64 @@ where
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::{cell::Cell, rc::Rc};
+
+    use super::*;
+
+    #[derive(Debug, Default, Clone)]
+    struct Srv(Rc<Cell<usize>>);
+
+    impl Service<()> for Srv {
+        type Response = ();
+        type Error = ();
+
+        async fn ready(&self, _: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
+            Ok(())
+        }
+
+        async fn call(&self, _: (), _: ServiceCtx<'_, Self>) -> Result<(), ()> {
+            Ok(())
+        }
+
+        async fn shutdown(&self) {
+            self.0.set(self.0.get() + 1);
+        }
+    }
+
+    #[ntex::test]
+    async fn pipeline_service() {
+        let cnt_sht = Rc::new(Cell::new(0));
+        let srv = Pipeline::new(
+            Pipeline::new(Srv(cnt_sht.clone()).map(|_| "ok"))
+                .into_service()
+                .clone(),
+        );
+        let res = srv.call(()).await;
+        assert!(res.is_ok());
+        assert_eq!(res.unwrap(), "ok");
+
+        let res = srv.ready().await;
+        assert_eq!(res, Ok(()));
+
+        srv.shutdown().await;
+        assert_eq!(cnt_sht.get(), 1);
+        let _ = format!("{srv:?}");
+
+        let cnt_sht = Rc::new(Cell::new(0));
+        let svc = Srv(cnt_sht.clone()).map(|_| "ok");
+        let srv = Pipeline::new(PipelineSvc::from(&svc));
+        let res = srv.call(()).await;
+        assert!(res.is_ok());
+        assert_eq!(res.unwrap(), "ok");
+
+        let res = srv.ready().await;
+        assert_eq!(res, Ok(()));
+
+        srv.shutdown().await;
+        assert_eq!(cnt_sht.get(), 1);
+        let _ = format!("{srv:?}");
+    }
+}

ntex/Cargo.toml

@@ -70,7 +70,7 @@ no-test-logging = []
 ntex-codec = "1"
 ntex-http = "1"
 ntex-router = "1"
-ntex-service = "3.7.3"
+ntex-service = "3.8.0"
 ntex-macros = "0.2"
 ntex-util = "3.3"
 ntex-bytes = "1"