From f3ce9388251303bc307f43871a93fa0763ef4c8b Mon Sep 17 00:00:00 2001 From: Saurav Date: Mon, 22 Jun 2026 12:01:40 +0000 Subject: [PATCH] feat(grpc): add ServerBuilder, registration, interceptors, and routing Introduce the server-side handle/router API for building a gRPC `Server` from a fluent builder. - ServerBuilder: fluent construction via `Server::builder()`, with `add_service`, `interceptor`and `build` / `build_with_runtime`. - Service + ServiceExt: `Service` trait for method registration, plus `with_interceptor` to wrap all of a service's methods (InterceptedService). - Interceptors: `Intercept` trait, no-op `Identity`, and `InterceptExt::chain` for composing interceptors into an `InterceptorChain` (first added runs outermost). - Descriptors: `ServiceDescriptor`, `MethodDescriptor`, and `MethodType`. - Routing: `RouterBuilder` maps method paths to `DynHandle`s. - Options: `ServerOptions` currently empty, but a kitchen sink for options. --- grpc/src/server/builder.rs | 308 ++++++++++++++ grpc/src/server/descriptor.rs | 181 ++++++++ grpc/src/server/interceptor.rs | 295 +++++++++++++ grpc/src/server/mod.rs | 112 +++-- grpc/src/server/options.rs | 45 ++ grpc/src/server/router.rs | 733 +++++++++++++++++++++++++++++++++ grpc/src/server/service.rs | 325 +++++++++++++++ 7 files changed, 1976 insertions(+), 23 deletions(-) create mode 100644 grpc/src/server/builder.rs create mode 100644 grpc/src/server/descriptor.rs create mode 100644 grpc/src/server/options.rs create mode 100644 grpc/src/server/router.rs create mode 100644 grpc/src/server/service.rs diff --git a/grpc/src/server/builder.rs b/grpc/src/server/builder.rs new file mode 100644 index 000000000..4c9a921f0 --- /dev/null +++ b/grpc/src/server/builder.rs @@ -0,0 +1,308 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use crate::rt::GrpcRuntime; +use crate::server::Server; +use crate::server::ServerOptions; +use crate::server::interceptor::Identity; +use crate::server::interceptor::Intercept; +use crate::server::interceptor::InterceptorChain; +use crate::server::router::RouterBuilder; +use crate::server::service::Service; + +/// A fluent builder for constructing a [`Server`]. +/// +/// Register services with [`add_service()`](ServerBuilder::add_service) and add +/// global interceptors with [`interceptor()`](ServerBuilder::interceptor), then +/// finish with [`build()`](ServerBuilder::build). +/// +/// # Examples +/// +/// ```ignore +/// let server = Server::builder() +/// .interceptor(auth) +/// .add_service(greeter_service) +/// .build(); +/// ``` +pub struct ServerBuilder { + router: RouterBuilder, + options: ServerOptions, +} + +// --------------------------------------------------------------------------- +// Constructor +// --------------------------------------------------------------------------- + +impl ServerBuilder { + /// Creates a new `ServerBuilder` with no interceptors. + pub fn new() -> Self { + ServerBuilder { + router: RouterBuilder::new(), + options: ServerOptions::default(), + } + } +} + +impl Default for ServerBuilder { + fn default() -> Self { + Self::new() + } +} + +// --------------------------------------------------------------------------- +// Setters — available on any ServerBuilder +// --------------------------------------------------------------------------- + +impl ServerBuilder { + /// Adds a global interceptor applied to every registered method. + /// + /// May be called repeatedly to compose multiple interceptors. + pub fn interceptor(self, next: J) -> ServerBuilder> + where + J: Intercept + Clone + Send + Sync + 'static, + { + ServerBuilder { + router: self.router.chain_interceptor(next), + options: self.options, + } + } + + /// Registers all methods from a [`Service`]. + pub fn add_service(mut self, service: impl Service) -> Self { + self.router = self.router.add_service(service); + self + } + + /// Builds the [`Server`] with the explicitly provided runtime. + /// + /// Always available. When the `_runtime-tokio` feature is enabled, + /// [`build()`](ServerBuilder::build) can be used instead to use the default + /// Tokio runtime. + pub fn build_with_runtime(self, runtime: GrpcRuntime) -> Server { + Server::new(self.router.build(), runtime, self.options) + } +} + +// --------------------------------------------------------------------------- +// build() — uses the default runtime when _runtime-tokio is enabled +// --------------------------------------------------------------------------- + +#[cfg(feature = "_runtime-tokio")] +impl ServerBuilder { + /// Builds the [`Server`] using the default Tokio runtime. + /// + /// Available only when the `_runtime-tokio` feature is enabled. Without it, + /// use [`build_with_runtime()`](ServerBuilder::build_with_runtime). + pub fn build(self) -> Server { + Server::new( + self.router.build(), + crate::rt::default_runtime(), + self.options, + ) + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use tokio::sync::Mutex; + + use crate::client::CallOptions; + use crate::core::RecvMessage; + use crate::server::DynHandle; + use crate::server::Handle; + use crate::server::RecvStream; + use crate::server::RequestHeaders; + use crate::server::ResponseStreamItem; + use crate::server::SendOptions; + use crate::server::SendStream; + use crate::server::Server; + use crate::server::Trailers; + use crate::server::descriptor::{MethodDescriptor, MethodType, ServiceDescriptor}; + use crate::server::interceptor::{Intercept, InterceptExt}; + use crate::server::service::Service; + + struct MockSendStream; + impl SendStream for MockSendStream { + async fn send<'a>( + &mut self, + _item: ResponseStreamItem<'a>, + _options: SendOptions, + ) -> Result<(), ()> { + Ok(()) + } + } + + struct MockRecvStream; + impl RecvStream for MockRecvStream { + async fn next(&mut self, _msg: &mut dyn RecvMessage) -> Option> { + None + } + } + + struct TrackingHandler { + order: Arc>>, + } + + impl Handle for TrackingHandler { + async fn handle( + &self, + _headers: RequestHeaders, + _options: CallOptions, + _tx: &mut impl SendStream, + _rx: impl RecvStream + 'static, + ) -> Trailers { + self.order.lock().await.push(0); + Trailers::new(Ok(())) + } + } + + #[derive(Clone)] + struct TrackingInterceptor { + id: usize, + order: Arc>>, + } + + impl Intercept for TrackingInterceptor { + async fn intercept( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + next: &impl Handle, + ) -> Trailers { + self.order.lock().await.push(self.id); + next.handle(headers, options, tx, rx).await + } + } + + struct TestService { + order: Arc>>, + } + + impl Service for TestService { + fn descriptor(&self) -> ServiceDescriptor { + ServiceDescriptor::new( + "test.Svc", + vec![MethodDescriptor::new("/test.Svc/Method", MethodType::Unary)], + ) + } + + fn register_methods(self) -> Vec<(String, Arc)> { + vec![( + "/test.Svc/Method".to_string(), + Arc::new(TrackingHandler { order: self.order }), + )] + } + } + + #[test] + fn server_builder_builds_without_services() { + // Should not panic — creates a server with empty router. + let _server = Server::builder().build(); + } + + #[tokio::test] + async fn server_builder_with_interceptor_chain_and_service() { + let order = Arc::new(Mutex::new(Vec::new())); + let int_a = TrackingInterceptor { + id: 1, + order: order.clone(), + }; + let int_b = TrackingInterceptor { + id: 2, + order: order.clone(), + }; + let chain = int_a.chain(int_b); + + let svc = TestService { + order: order.clone(), + }; + + let _server = Server::builder() + .interceptor(chain) + .add_service(svc) + .build(); + } + + #[tokio::test] + async fn server_builder_interceptor_method() { + let order = Arc::new(Mutex::new(Vec::new())); + let int_a = TrackingInterceptor { + id: 1, + order: order.clone(), + }; + let int_b = TrackingInterceptor { + id: 2, + order: order.clone(), + }; + let chain = int_a.chain(int_b); + + let svc = TestService { + order: order.clone(), + }; + + // Test the new .interceptor() API. + let _server = Server::builder() + .interceptor(chain) + .add_service(svc) + .build(); + } + + #[test] + fn server_builder_multiple_services() { + struct SvcA; + impl Service for SvcA { + fn descriptor(&self) -> ServiceDescriptor { + ServiceDescriptor::new("test.A", vec![]) + } + fn register_methods(self) -> Vec<(String, Arc)> { + vec![] + } + } + + struct SvcB; + impl Service for SvcB { + fn descriptor(&self) -> ServiceDescriptor { + ServiceDescriptor::new("test.B", vec![]) + } + fn register_methods(self) -> Vec<(String, Arc)> { + vec![] + } + } + + let _server = Server::builder() + .add_service(SvcA) + .add_service(SvcB) + .build(); + } + + #[test] + fn server_builder_with_explicit_runtime() { + let rt = crate::rt::default_runtime(); + let _server = Server::builder().build_with_runtime(rt); + } +} diff --git a/grpc/src/server/descriptor.rs b/grpc/src/server/descriptor.rs new file mode 100644 index 000000000..a9db556df --- /dev/null +++ b/grpc/src/server/descriptor.rs @@ -0,0 +1,181 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +/// The type (cardinality) of a gRPC method. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MethodType { + /// One request message followed by one response message. + Unary, + /// Zero or more request messages with one response message. + ClientStreaming, + /// One request message followed by zero or more response messages. + ServerStreaming, + /// Zero or more request and response messages arbitrarily interleaved. + BidiStreaming, +} + +/// Pure metadata about a single gRPC method. +/// +/// This is a data class — it carries no handler logic. It describes what a +/// method looks like (its path and cardinality) without specifying how it's +/// implemented. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub struct MethodDescriptor { + /// Full method path, e.g., `"/helloworld.Greeter/SayHello"`. + full_path: String, + /// The method cardinality. + method_type: MethodType, +} + +impl MethodDescriptor { + /// Creates a descriptor for the given method path and cardinality. + pub fn new(full_path: impl Into, method_type: MethodType) -> Self { + Self { + full_path: full_path.into(), + method_type, + } + } + + /// Returns the full method path, e.g., `"/helloworld.Greeter/SayHello"`. + pub fn full_path(&self) -> &str { + &self.full_path + } + + /// Consumes the descriptor, returning its owned full method path. + pub fn into_full_path(self) -> String { + self.full_path + } + + /// Returns the method cardinality. + pub fn method_type(&self) -> MethodType { + self.method_type + } +} + +/// Pure metadata about a gRPC service. +/// +/// This is a data class — it carries no handler logic. It describes what a +/// service looks like (its name and the methods it contains) without +/// specifying how they're implemented. +#[derive(Debug, Clone)] +#[non_exhaustive] +pub struct ServiceDescriptor { + /// Fully qualified service name, e.g., `"helloworld.Greeter"`. + name: String, + /// Descriptors for all methods in this service. + methods: Vec, +} + +impl ServiceDescriptor { + /// Creates a descriptor for the given service name and methods. + pub fn new(name: impl Into, methods: Vec) -> Self { + Self { + name: name.into(), + methods, + } + } + + /// Returns the fully qualified service name, e.g., `"helloworld.Greeter"`. + pub fn name(&self) -> &str { + &self.name + } + + /// Returns the descriptors for all methods in this service. + pub fn methods(&self) -> &[MethodDescriptor] { + &self.methods + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn method_type_variants() { + assert_eq!(MethodType::Unary, MethodType::Unary); + assert_ne!(MethodType::Unary, MethodType::ClientStreaming); + assert_ne!(MethodType::ServerStreaming, MethodType::BidiStreaming); + } + + #[test] + fn method_descriptor_exposes_path_and_type() { + let desc = MethodDescriptor::new("/pkg.Svc/Method", MethodType::Unary); + assert_eq!(desc.full_path(), "/pkg.Svc/Method"); + assert_eq!(desc.method_type(), MethodType::Unary); + } + + #[test] + fn method_descriptor_new_accepts_string() { + let desc = + MethodDescriptor::new("/pkg.Svc/Method".to_string(), MethodType::ServerStreaming); + assert_eq!(desc.full_path(), "/pkg.Svc/Method"); + assert_eq!(desc.method_type(), MethodType::ServerStreaming); + } + + #[test] + fn method_descriptor_into_full_path() { + let desc = MethodDescriptor::new("/pkg.Svc/Method", MethodType::Unary); + assert_eq!(desc.into_full_path(), "/pkg.Svc/Method"); + } + + #[test] + fn method_descriptor_clone() { + let desc = MethodDescriptor::new("/pkg.Svc/Method", MethodType::BidiStreaming); + let cloned = desc.clone(); + assert_eq!(cloned.full_path(), desc.full_path()); + assert_eq!(cloned.method_type(), desc.method_type()); + } + + #[test] + fn service_descriptor_exposes_name_and_methods() { + let desc = ServiceDescriptor::new( + "pkg.Svc", + vec![ + MethodDescriptor::new("/pkg.Svc/M1", MethodType::Unary), + MethodDescriptor::new("/pkg.Svc/M2", MethodType::ServerStreaming), + ], + ); + assert_eq!(desc.name(), "pkg.Svc"); + assert_eq!(desc.methods().len(), 2); + assert_eq!(desc.methods()[0].full_path(), "/pkg.Svc/M1"); + } + + #[test] + fn service_descriptor_empty_methods() { + let desc = ServiceDescriptor::new("pkg.Empty", vec![]); + assert_eq!(desc.methods().len(), 0); + } + + #[test] + fn service_descriptor_clone() { + let desc = ServiceDescriptor::new( + "pkg.Svc", + vec![MethodDescriptor::new("/pkg.Svc/M1", MethodType::Unary)], + ); + let cloned = desc.clone(); + assert_eq!(cloned.name(), desc.name()); + assert_eq!(cloned.methods().len(), desc.methods().len()); + } +} diff --git a/grpc/src/server/interceptor.rs b/grpc/src/server/interceptor.rs index 20c9d4a5b..be6061f69 100644 --- a/grpc/src/server/interceptor.rs +++ b/grpc/src/server/interceptor.rs @@ -89,6 +89,114 @@ pub trait HandleExt: Handle + Sized { impl HandleExt for T {} +/// A no-op interceptor that simply delegates to the next handler. +/// +/// This is the default interceptor used by [`RouterBuilder`](crate::server::RouterBuilder) +/// when no interceptor has been added. +#[derive(Clone, Copy)] +pub struct Identity; + +impl Intercept for Identity { + async fn intercept( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + next: &impl Handle, + ) -> Trailers { + next.handle(headers, options, tx, rx).await + } +} + +/// Extension trait for chaining [`Intercept`] implementations. +/// +/// Provides the [`chain`](InterceptExt::chain) method, which composes two +/// interceptors into a single [`InterceptorChain`] that itself implements +/// `Intercept`. Chains nest naturally via repeated calls: +/// +/// ```ignore +/// let pipeline = logging.chain(auth).chain(rate_limit); +/// // execution order: logging → auth → rate_limit → handler +/// ``` +pub trait InterceptExt: Intercept + Sized { + /// Chains `self` with `next`, returning an [`InterceptorChain`] where + /// `self` runs first and `next` runs second. + fn chain(self, next: I) -> InterceptorChain { + InterceptorChain { + first: self, + second: next, + } + } +} + +impl InterceptExt for T {} + +/// Two interceptors chained together, where `first` runs before `second`. +/// +/// Created via [`InterceptExt::chain`]. Itself implements [`Intercept`], so +/// chains compose recursively: +/// `InterceptorChain>` runs A → B → C. +/// +/// ```ignore +/// let pipeline = logging.chain(auth); +/// server.interceptor(pipeline); +/// ``` +#[derive(Clone)] +pub struct InterceptorChain { + first: A, + second: B, +} + +impl Intercept for InterceptorChain +where + A: Intercept, + B: Intercept, +{ + async fn intercept( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + next: &impl Handle, + ) -> Trailers { + // Build a temporary Handle that runs `second` then delegates to `next`. + let inner = SecondThenNext { + second: &self.second, + next, + }; + self.first.intercept(headers, options, tx, rx, &inner).await + } +} + +/// A temporary Handle adapter used inside [`InterceptorChain`]. +/// +/// When called, it runs `second.intercept(...)` with `next` as the +/// downstream handler, achieving the chained execution order. +struct SecondThenNext<'a, B, N: ?Sized> { + second: &'a B, + next: &'a N, +} + +impl Handle for SecondThenNext<'_, B, N> +where + B: Intercept, + N: Handle, +{ + async fn handle( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + ) -> Trailers { + self.second + .intercept(headers, options, tx, rx, self.next) + .await + } +} + #[cfg(test)] mod test { use std::sync::Arc; @@ -101,6 +209,8 @@ mod test { use crate::server::RequestHeaders; use crate::server::ResponseStreamItem; use crate::server::SendOptions; + use crate::status::StatusCodeError; + use crate::status::StatusError; struct MockSendStream; impl SendStream for MockSendStream { @@ -261,4 +371,189 @@ mod test { let final_order = order.lock().await; assert_eq!(*final_order, vec![1, 2, 0]); } + + // --- Chaining tests exercising `InterceptExt::chain` / `InterceptorChain` --- + + /// Records its `id` when run, then delegates to `next`. + struct TrackingInterceptor { + id: usize, + order: Arc>>, + } + + impl Intercept for TrackingInterceptor { + async fn intercept( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + next: &impl Handle, + ) -> Trailers { + self.order.lock().await.push(self.id); + next.handle(headers, options, tx, rx).await + } + } + + /// Records `id` and returns without calling `next` (short-circuit). + struct ShortCircuitInterceptor { + id: usize, + order: Arc>>, + } + + impl Intercept for ShortCircuitInterceptor { + async fn intercept( + &self, + _headers: RequestHeaders, + _options: CallOptions, + _tx: &mut impl SendStream, + _rx: impl RecvStream + 'static, + _next: &impl Handle, + ) -> Trailers { + self.order.lock().await.push(self.id); + Trailers::new(Ok(())) + } + } + + /// Records `0` (handler marker) and returns the given trailers. + struct TrackingHandler { + order: Arc>>, + trailers: Trailers, + } + + impl Handle for TrackingHandler { + async fn handle( + &self, + _headers: RequestHeaders, + _options: CallOptions, + _tx: &mut impl SendStream, + _rx: impl RecvStream + 'static, + ) -> Trailers { + self.order.lock().await.push(0); + self.trailers.clone() + } + } + + fn tracking_interceptor(id: usize, order: &Arc>>) -> TrackingInterceptor { + TrackingInterceptor { + id, + order: order.clone(), + } + } + + /// Invokes `chain.intercept(...)` against a `TrackingHandler` and returns the + /// resulting `Trailers`. + async fn run_chain(chain: &impl Intercept, handler: &impl Handle) -> Trailers { + let mut tx = MockSendStream; + let rx = MockRecvStream; + chain + .intercept( + RequestHeaders::default(), + CallOptions::default(), + &mut tx, + rx, + handler, + ) + .await + } + + #[tokio::test] + async fn chain_runs_first_then_second_then_handler() { + let order = Arc::new(Mutex::new(Vec::new())); + let chain = tracking_interceptor(1, &order).chain(tracking_interceptor(2, &order)); + let handler = TrackingHandler { + order: order.clone(), + trailers: Trailers::new(Ok(())), + }; + + run_chain(&chain, &handler).await; + + assert_eq!(*order.lock().await, vec![1, 2, 0]); + } + + #[tokio::test] + async fn nested_chain_flattens_left_to_right() { + let order = Arc::new(Mutex::new(Vec::new())); + let chain = tracking_interceptor(1, &order) + .chain(tracking_interceptor(2, &order)) + .chain(tracking_interceptor(3, &order)); + let handler = TrackingHandler { + order: order.clone(), + trailers: Trailers::new(Ok(())), + }; + + run_chain(&chain, &handler).await; + + assert_eq!(*order.lock().await, vec![1, 2, 3, 0]); + } + + #[tokio::test] + async fn chain_of_chains() { + let order = Arc::new(Mutex::new(Vec::new())); + let left = tracking_interceptor(1, &order).chain(tracking_interceptor(2, &order)); + let right = tracking_interceptor(3, &order).chain(tracking_interceptor(4, &order)); + let chain = left.chain(right); + let handler = TrackingHandler { + order: order.clone(), + trailers: Trailers::new(Ok(())), + }; + + run_chain(&chain, &handler).await; + + assert_eq!(*order.lock().await, vec![1, 2, 3, 4, 0]); + } + + #[tokio::test] + async fn identity_is_transparent() { + let order = Arc::new(Mutex::new(Vec::new())); + let chain = Identity.chain(tracking_interceptor(1, &order)); + let handler = TrackingHandler { + order: order.clone(), + trailers: Trailers::new(Ok(())), + }; + + run_chain(&chain, &handler).await; + + // Identity adds no entry; only interceptor 1 then the handler run. + assert_eq!(*order.lock().await, vec![1, 0]); + } + + #[tokio::test] + async fn interceptor_can_short_circuit() { + let order = Arc::new(Mutex::new(Vec::new())); + let chain = ShortCircuitInterceptor { + id: 1, + order: order.clone(), + } + .chain(tracking_interceptor(2, &order)); + let handler = TrackingHandler { + order: order.clone(), + trailers: Trailers::new(Ok(())), + }; + + run_chain(&chain, &handler).await; + + // The short-circuiting interceptor never calls `next`, so neither the + // downstream interceptor (2) nor the handler (0) run. + assert_eq!(*order.lock().await, vec![1]); + } + + #[tokio::test] + async fn trailers_propagate_back_through_chain() { + let order = Arc::new(Mutex::new(Vec::new())); + let chain = tracking_interceptor(1, &order).chain(tracking_interceptor(2, &order)); + let distinctive = Trailers::new(Err(StatusError::new( + StatusCodeError::FailedPrecondition, + "from-handler", + ))); + let handler = TrackingHandler { + order: order.clone(), + trailers: distinctive, + }; + + let result = run_chain(&chain, &handler).await; + + let err = result.into_status().unwrap_err(); + assert_eq!(err.code(), StatusCodeError::FailedPrecondition); + assert_eq!(err.message(), "from-handler"); + } } diff --git a/grpc/src/server/mod.rs b/grpc/src/server/mod.rs index c54ef6f4d..c1111db15 100644 --- a/grpc/src/server/mod.rs +++ b/grpc/src/server/mod.rs @@ -57,7 +57,14 @@ use crate::core::SendMessage; use crate::metadata::MetadataMap; use crate::rt::GrpcRuntime; +pub mod builder; +pub mod descriptor; pub(crate) mod interceptor; +pub mod options; +pub(crate) mod router; +pub mod service; + +pub use options::ServerOptions; /// A serving connection that supports graceful shutdown. /// @@ -82,6 +89,7 @@ pub trait GracefulConnection: Future + Send + 'static { pub struct Server { handler: Option>, runtime: GrpcRuntime, + options: ServerOptions, } mod sealed { @@ -173,20 +181,40 @@ impl GracefulCoordinator { } impl Server { - /// Creates a new server with no handler. - pub fn new() -> Self { + /// Creates a [`ServerBuilder`](builder::ServerBuilder) with no interceptors. + /// + /// # Example + /// + /// ```ignore + /// let server = Server::builder() + /// .add_service(greeter_service) + /// .build(); + /// ``` + pub fn builder() -> builder::ServerBuilder { + builder::ServerBuilder::new() + } + + /// Creates a new server with the given handler, runtime, and options. + pub(crate) fn new( + handler: impl Handle + 'static, + runtime: GrpcRuntime, + options: ServerOptions, + ) -> Self { Self { - handler: None, - runtime: crate::rt::default_runtime(), + handler: Some(Arc::new(handler)), + runtime, + options, } } - /// Sets the RPC handler for this server. - pub fn set_handler(&mut self, h: H) - where - H: Handle + Send + Sync + 'static, - { - self.handler = Some(Arc::new(h)) + /// Returns the runtime used by this server. + pub fn runtime(&self) -> &GrpcRuntime { + &self.runtime + } + + /// Returns the server options. + pub fn options(&self) -> &ServerOptions { + &self.options } /// Serves on the given listener until it stops producing connections. @@ -256,7 +284,11 @@ impl Server { impl Default for Server { fn default() -> Self { - Self::new() + Self { + handler: None, + runtime: crate::rt::default_runtime(), + options: ServerOptions::default(), + } } } @@ -702,7 +734,7 @@ mod tests { #[tokio::test] async fn server_stops_on_shutdown_signal() { let listener = crate::inmemory::InMemoryListener::new(); - let server = Server::new(); + let server = Server::builder().build(); let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); @@ -728,7 +760,7 @@ mod tests { #[tokio::test] async fn server_stops_when_listener_closes() { let listener = crate::inmemory::InMemoryListener::new(); - let server = Server::new(); + let server = Server::builder().build(); let listener_for_serve = listener.clone(); let server_handle = tokio::spawn(async move { @@ -772,7 +804,7 @@ mod tests { #[tokio::test] async fn dropping_serve_future_force_closes_connection() { let listener = crate::inmemory::InMemoryListener::new(); - let server = Server::new(); + let server = Server::builder().build(); // A never-resolving signal future (we won't signal gracefully, we will drop the serve future directly). let (_signal_tx, signal_rx) = tokio::sync::oneshot::channel::<()>(); @@ -931,7 +963,12 @@ mod tests { let mut tx = NopSendStream; let rx = BoxedRecvStream(Box::new(NopRecvStream)); let _ = handler - .dyn_handle(RequestHeaders::new(), CallOptions::new(), &mut tx, rx) + .dyn_handle( + RequestHeaders::new().with_method_name("/test.Draining/Method"), + CallOptions::new(), + &mut tx, + rx, + ) .await; }); MockServingConnection { inner } @@ -941,12 +978,12 @@ mod tests { #[tokio::test] async fn listener_dropped_when_shutdown_signal_fires() { let (listener, dropped, _tx) = MockListener::new(); - let server = Server::new(); let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); let server_handle = tokio::spawn(async move { - server + Server::builder() + .build() .serve_with_shutdown(listener, async { let _ = shutdown_rx.await; }) @@ -974,13 +1011,13 @@ mod tests { #[tokio::test] async fn listener_dropped_immediately_while_connections_drain() { use crate::client::CallOptions; + use crate::server::descriptor::{MethodDescriptor, MethodType, ServiceDescriptor}; + use crate::server::service::Service; use crate::server::{RecvStream, SendStream}; use crate::server::{RequestHeaders, Trailers}; let (listener, dropped, tx) = MockListener::new(); - let mut server = Server::new(); - let handler_started = Arc::new(AtomicBool::new(false)); let (unblock_tx, unblock_rx) = tokio::sync::oneshot::channel::<()>(); let unblock_rx = Arc::new(tokio::sync::Mutex::new(Some(unblock_rx))); @@ -1006,10 +1043,39 @@ mod tests { } } - server.set_handler(DrainingHandler { - started: handler_started.clone(), - unblock: unblock_rx, - }); + struct DrainingService { + started: Arc, + unblock: Arc>>>, + } + + impl Service for DrainingService { + fn descriptor(&self) -> ServiceDescriptor { + ServiceDescriptor::new( + "test.Draining", + vec![MethodDescriptor::new( + "/test.Draining/Method", + MethodType::Unary, + )], + ) + } + + fn register_methods(self) -> Vec<(String, Arc)> { + vec![( + "/test.Draining/Method".to_string(), + Arc::new(DrainingHandler { + started: self.started, + unblock: self.unblock, + }), + )] + } + } + + let server = Server::builder() + .add_service(DrainingService { + started: handler_started.clone(), + unblock: unblock_rx, + }) + .build(); let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); diff --git a/grpc/src/server/options.rs b/grpc/src/server/options.rs new file mode 100644 index 000000000..694bb485d --- /dev/null +++ b/grpc/src/server/options.rs @@ -0,0 +1,45 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +/// A read-only snapshot of the server-wide options that apply across all +/// connections. +/// +/// Options are configured through [`ServerBuilder`](crate::server::builder::ServerBuilder), +/// which is the single mutation surface. This type is the read side: obtain it +/// via [`Server::options()`](crate::server::Server::options) to introspect the +/// effective configuration a server is running with. +/// +/// # Planned options +/// +/// This type is currently empty because no server-wide options are enforced +/// yet. Options that have been considered but are not implemented: +/// +/// - **Maximum concurrent RPCs** (`max_concurrent_rpcs`): a cap on the number +/// of in-flight RPCs across all connections, beyond which new RPCs would be +/// rejected with `RESOURCE_EXHAUSTED`. This was previously exposed on the +/// builder but removed because nothing enforced it (it was a no-op). It +/// should be reintroduced alongside the machinery that actually applies the +/// limit. +#[derive(Debug, Clone, Default)] +pub struct ServerOptions {} diff --git a/grpc/src/server/router.rs b/grpc/src/server/router.rs new file mode 100644 index 000000000..d91ecba20 --- /dev/null +++ b/grpc/src/server/router.rs @@ -0,0 +1,733 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use std::collections::HashMap; +use std::sync::Arc; + +use crate::StatusCodeError; +use crate::StatusError; +use crate::client::CallOptions; +use crate::server::BoxedRecvStream; +use crate::server::DynHandle; +use crate::server::DynRecvStream; +use crate::server::DynSendStream; +use crate::server::Handle; +use crate::server::RecvStream; +use crate::server::RequestHeaders; +use crate::server::SendStream; +use crate::server::Trailers; +use crate::server::descriptor::{MethodDescriptor, ServiceDescriptor}; +use crate::server::interceptor::Identity; +use crate::server::interceptor::Intercept; +use crate::server::interceptor::InterceptExt; +use crate::server::interceptor::InterceptorChain; +use crate::server::service::Service; + +/// A builder for constructing an immutable [`Router`]. +pub(crate) struct RouterBuilder { + handlers: HashMap>, + descriptors: Vec, + interceptor: I, +} + +impl RouterBuilder { + /// Creates a new, empty `RouterBuilder` with no interceptors. + pub(crate) fn new() -> RouterBuilder { + RouterBuilder { + handlers: HashMap::new(), + descriptors: Vec::new(), + interceptor: Identity, + } + } +} + +impl RouterBuilder +where + I: Intercept + Clone + Send + Sync + 'static, +{ + /// Chains an additional interceptor after the existing stack. + pub(crate) fn chain_interceptor(self, next: J) -> RouterBuilder> + where + J: Intercept + Clone + Send + Sync + 'static, + { + RouterBuilder { + handlers: self.handlers, + descriptors: self.descriptors, + interceptor: self.interceptor.chain(next), + } + } + + /// Registers `handler` for the given method (last-one-wins on duplicate paths). + pub(crate) fn add_method(mut self, descriptor: MethodDescriptor, handler: H) -> Self + where + H: Handle + Send + Sync + 'static, + { + self.handlers + .insert(descriptor.into_full_path(), Arc::new(handler)); + self + } + + /// Registers all methods from a [`Service`]. + pub(crate) fn add_service(mut self, service: impl Service) -> Self { + self.descriptors.push(service.descriptor()); + for (path, handler) in service.register_methods() { + self.handlers.insert(path, handler); + } + self + } + + /// Consumes this builder and produces an immutable [`Router`]. + pub(crate) fn build(self) -> Router { + Router { + handlers: self.handlers, + interceptor: self.interceptor, + } + } +} + +impl Default for RouterBuilder { + fn default() -> Self { + Self::new() + } +} + +/// Routes incoming gRPC RPCs to the correct handler based on the request's +/// method name. +/// +/// `Router` implements [`Handle`], so it can be used as the handler for a +/// [`Server`](crate::server::Server). For most use cases, prefer +/// [`Server::builder()`](crate::server::Server::builder) which constructs +/// both the router and server together. +/// +/// A `Router` is immutable once built; use [`RouterBuilder`] to construct one. +/// +/// # Example +/// +/// ```ignore +/// let server = Server::builder() +/// .add_service(my_service) +/// .build(); +/// ``` +pub struct Router { + handlers: HashMap>, + interceptor: I, +} + +impl Handle for Router +where + I: Intercept + Send + Sync + 'static, +{ + async fn handle( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + ) -> Trailers { + // Resolve the method first: unknown methods short-circuit to + // UNIMPLEMENTED without running the interceptor stack. + let Some(handler) = self.handlers.get(headers.method_name()) else { + return Trailers::new(Err(StatusError::new( + StatusCodeError::Unimplemented, + format!("unknown method: {}", headers.method_name()), + ))); + }; + + // Apply the global interceptor stack around the resolved handler. + let next = DynHandleRef(handler.as_ref()); + self.interceptor + .intercept(headers, options, tx, rx, &next) + .await + } +} + +/// Adapts a type-erased [`DynHandle`] back into a [`Handle`] so the global +/// interceptor stack can invoke it as its `next` handler. +/// +/// The only dynamic dispatch is the leaf `dyn_handle` call; the interceptor +/// stack itself remains monomorphized. +struct DynHandleRef<'a>(&'a dyn DynHandle); + +impl Handle for DynHandleRef<'_> { + async fn handle( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + ) -> Trailers { + // Bridge from `impl SendStream` → `&mut dyn DynSendStream` and + // `impl RecvStream` → `BoxedRecvStream`, matching the blanket + // `DynHandle for T: Handle` pattern. + let mut dyn_tx: &mut dyn DynSendStream = tx; + let boxed_rx = BoxedRecvStream(Box::new(rx) as Box); + self.0 + .dyn_handle(headers, options, &mut dyn_tx, boxed_rx) + .await + } +} + +#[cfg(test)] +mod test { + use std::sync::Arc; + + use tokio::sync::Mutex; + + use super::*; + use crate::client::CallOptions; + use crate::core::RecvMessage; + use crate::server::RequestHeaders; + use crate::server::ResponseStreamItem; + use crate::server::SendOptions; + use crate::server::descriptor::{MethodDescriptor, MethodType, ServiceDescriptor}; + + struct MockSendStream; + impl SendStream for MockSendStream { + async fn send<'a>( + &mut self, + _item: ResponseStreamItem<'a>, + _options: SendOptions, + ) -> Result<(), ()> { + Ok(()) + } + } + + struct MockRecvStream; + impl RecvStream for MockRecvStream { + async fn next(&mut self, _msg: &mut dyn RecvMessage) -> Option> { + None + } + } + + /// A handler that records its method name when called and returns OK. + struct RecordingHandler { + called_with: Arc>>, + } + + impl Handle for RecordingHandler { + async fn handle( + &self, + headers: RequestHeaders, + _options: CallOptions, + _tx: &mut impl SendStream, + _rx: impl RecvStream + 'static, + ) -> Trailers { + let mut called = self.called_with.lock().await; + *called = Some(headers.method_name().to_string()); + Trailers::new(Ok(())) + } + } + + #[tokio::test] + async fn test_registered_method_dispatches() { + let called_with = Arc::new(Mutex::new(None)); + let handler = RecordingHandler { + called_with: called_with.clone(), + }; + + let router = RouterBuilder::new() + .add_method( + MethodDescriptor::new("/pkg.Svc/Method", MethodType::Unary), + handler, + ) + .build(); + + let headers = RequestHeaders::new().with_method_name("/pkg.Svc/Method"); + + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + assert_eq!( + *called_with.lock().await, + Some("/pkg.Svc/Method".to_string()) + ); + } + + #[tokio::test] + async fn test_unregistered_method_returns_unimplemented() { + let router = RouterBuilder::new().build(); + + let headers = RequestHeaders::new().with_method_name("/pkg.Svc/NoSuchMethod"); + + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + let err = trailers.status().as_ref().unwrap_err(); + assert_eq!(err.code(), StatusCodeError::Unimplemented); + assert!( + err.message().contains("/pkg.Svc/NoSuchMethod"), + "error message should contain the method name, got: {}", + err.message() + ); + } + + #[tokio::test] + async fn test_last_one_wins_for_duplicate_paths() { + let first_called = Arc::new(Mutex::new(None)); + let second_called = Arc::new(Mutex::new(None)); + + let first_handler = RecordingHandler { + called_with: first_called.clone(), + }; + let second_handler = RecordingHandler { + called_with: second_called.clone(), + }; + + let router = RouterBuilder::new() + .add_method( + MethodDescriptor::new("/pkg.Svc/Method", MethodType::Unary), + first_handler, + ) + .add_method( + MethodDescriptor::new("/pkg.Svc/Method", MethodType::Unary), + second_handler, + ) + .build(); + + let headers = RequestHeaders::new().with_method_name("/pkg.Svc/Method"); + + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // The second handler should have been called (last-one-wins). + assert_eq!( + *second_called.lock().await, + Some("/pkg.Svc/Method".to_string()) + ); + // The first handler should NOT have been called. + assert!(first_called.lock().await.is_none()); + } + + #[tokio::test] + async fn test_add_service_visitor_pattern() { + let method_a_called = Arc::new(Mutex::new(None)); + let method_b_called = Arc::new(Mutex::new(None)); + + let handler_a = RecordingHandler { + called_with: method_a_called.clone(), + }; + let handler_b = RecordingHandler { + called_with: method_b_called.clone(), + }; + + struct MockService { + handler_a: RecordingHandler, + handler_b: RecordingHandler, + } + + impl Service for MockService { + fn descriptor(&self) -> ServiceDescriptor { + ServiceDescriptor::new( + "mock.MockService", + vec![ + MethodDescriptor::new("/mock.MockService/MethodA", MethodType::Unary), + MethodDescriptor::new("/mock.MockService/MethodB", MethodType::Unary), + ], + ) + } + + fn register_methods(self) -> Vec<(String, Arc)> { + vec![ + ( + "/mock.MockService/MethodA".to_string(), + Arc::new(self.handler_a), + ), + ( + "/mock.MockService/MethodB".to_string(), + Arc::new(self.handler_b), + ), + ] + } + } + + let service = MockService { + handler_a, + handler_b, + }; + + let router = RouterBuilder::new().add_service(service).build(); + + // Dispatch to MethodA. + let headers_a = RequestHeaders::new().with_method_name("/mock.MockService/MethodA"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + let trailers = router + .handle(headers_a, CallOptions::default(), &mut tx, rx) + .await; + assert!(trailers.status().is_ok()); + assert_eq!( + *method_a_called.lock().await, + Some("/mock.MockService/MethodA".to_string()) + ); + + // Dispatch to MethodB. + let headers_b = RequestHeaders::new().with_method_name("/mock.MockService/MethodB"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + let trailers = router + .handle(headers_b, CallOptions::default(), &mut tx, rx) + .await; + assert!(trailers.status().is_ok()); + assert_eq!( + *method_b_called.lock().await, + Some("/mock.MockService/MethodB".to_string()) + ); + } + + #[tokio::test] + async fn test_router_builder_default() { + let router = RouterBuilder::default().build(); + + let headers = RequestHeaders::new().with_method_name("/any.Service/AnyMethod"); + + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + let err = trailers.status().as_ref().unwrap_err(); + assert_eq!(err.code(), StatusCodeError::Unimplemented); + } + + // --- Interceptor-related test helpers --- + + use crate::server::interceptor::Intercept; + + /// An interceptor that pushes its `id` into a shared order vec, then + /// delegates to `next`. + #[derive(Clone)] + struct OrderInterceptor { + id: usize, + order: Arc>>, + } + + impl Intercept for OrderInterceptor { + async fn intercept( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + next: &impl Handle, + ) -> Trailers { + self.order.lock().await.push(self.id); + next.handle(headers, options, tx, rx).await + } + } + + /// A handler that pushes `0` into the shared order vec. + struct OrderHandler { + order: Arc>>, + } + + impl Handle for OrderHandler { + async fn handle( + &self, + _headers: RequestHeaders, + _options: CallOptions, + _tx: &mut impl SendStream, + _rx: impl RecvStream + 'static, + ) -> Trailers { + self.order.lock().await.push(0); + Trailers::new(Ok(())) + } + } + + #[tokio::test] + async fn test_router_builder_with_single_interceptor() { + let order = Arc::new(Mutex::new(Vec::new())); + + let interceptor = OrderInterceptor { + id: 1, + order: order.clone(), + }; + let handler = OrderHandler { + order: order.clone(), + }; + + let router = RouterBuilder::new() + .chain_interceptor(interceptor) + .add_method( + MethodDescriptor::new("/pkg.Svc/Method", MethodType::Unary), + handler, + ) + .build(); + + let headers = RequestHeaders::new().with_method_name("/pkg.Svc/Method"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // Interceptor (1) should run before handler (0). + assert_eq!(*order.lock().await, vec![1, 0]); + } + + #[tokio::test] + async fn test_router_builder_with_chained_interceptors() { + let order = Arc::new(Mutex::new(Vec::new())); + + let auth = OrderInterceptor { + id: 1, + order: order.clone(), + }; + let logging = OrderInterceptor { + id: 2, + order: order.clone(), + }; + let handler = OrderHandler { + order: order.clone(), + }; + + // First-added-first: auth (added first) should run before logging. + let router = RouterBuilder::new() + .chain_interceptor(auth) + .chain_interceptor(logging) + .add_method( + MethodDescriptor::new("/pkg.Svc/Method", MethodType::Unary), + handler, + ) + .build(); + + let headers = RequestHeaders::new().with_method_name("/pkg.Svc/Method"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // Execution order: auth(1) → logging(2) → handler(0). + assert_eq!(*order.lock().await, vec![1, 2, 0]); + } + + #[tokio::test] + async fn test_router_builder_interceptor_with_add_service() { + let order = Arc::new(Mutex::new(Vec::new())); + + let interceptor = OrderInterceptor { + id: 1, + order: order.clone(), + }; + + struct InterceptedService { + handler: OrderHandler, + } + + impl Service for InterceptedService { + fn descriptor(&self) -> ServiceDescriptor { + ServiceDescriptor::new( + "test.InterceptedService", + vec![MethodDescriptor::new( + "/test.InterceptedService/Method", + MethodType::Unary, + )], + ) + } + + fn register_methods(self) -> Vec<(String, Arc)> { + vec![( + "/test.InterceptedService/Method".to_string(), + Arc::new(self.handler), + )] + } + } + + let service = InterceptedService { + handler: OrderHandler { + order: order.clone(), + }, + }; + + let router = RouterBuilder::new() + .chain_interceptor(interceptor) + .add_service(service) + .build(); + + let headers = RequestHeaders::new().with_method_name("/test.InterceptedService/Method"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // Interceptor (1) should run before handler (0) even via add_service. + assert_eq!(*order.lock().await, vec![1, 0]); + } + + #[tokio::test] + async fn test_global_interceptor_skipped_for_unknown_method() { + let order = Arc::new(Mutex::new(Vec::new())); + + let interceptor = OrderInterceptor { + id: 1, + order: order.clone(), + }; + + // Global interceptor registered, but no handler for the requested method. + let router = RouterBuilder::new() + .chain_interceptor(interceptor) + .add_method( + MethodDescriptor::new("/pkg.Svc/Known", MethodType::Unary), + OrderHandler { + order: order.clone(), + }, + ) + .build(); + + let headers = RequestHeaders::new().with_method_name("/pkg.Svc/Unknown"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + // Unknown method short-circuits to UNIMPLEMENTED... + assert_eq!( + trailers.status().as_ref().unwrap_err().code(), + StatusCodeError::Unimplemented + ); + // ...without running the global interceptor (resolve-then-intercept). + assert!(order.lock().await.is_empty()); + } + + #[tokio::test] + async fn test_recallable_interceptor_accumulates() { + let order = Arc::new(Mutex::new(Vec::new())); + + let first = OrderInterceptor { + id: 1, + order: order.clone(), + }; + let second = OrderInterceptor { + id: 2, + order: order.clone(), + }; + let handler = OrderHandler { + order: order.clone(), + }; + + // Two separate chain_interceptor calls accumulate; first added runs first. + let router = RouterBuilder::new() + .chain_interceptor(first) + .chain_interceptor(second) + .add_method( + MethodDescriptor::new("/pkg.Svc/Method", MethodType::Unary), + handler, + ) + .build(); + + let headers = RequestHeaders::new().with_method_name("/pkg.Svc/Method"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // Execution order: first(1) → second(2) → handler(0). + assert_eq!(*order.lock().await, vec![1, 2, 0]); + } + + #[tokio::test] + async fn test_global_runs_before_per_service_interceptor() { + use crate::server::service::ServiceExt; + + let order = Arc::new(Mutex::new(Vec::new())); + + let global = OrderInterceptor { + id: 1, + order: order.clone(), + }; + let per_service = OrderInterceptor { + id: 2, + order: order.clone(), + }; + + struct SingleMethodService { + handler: OrderHandler, + } + + impl Service for SingleMethodService { + fn descriptor(&self) -> ServiceDescriptor { + ServiceDescriptor::new( + "test.Svc", + vec![MethodDescriptor::new("/test.Svc/Method", MethodType::Unary)], + ) + } + + fn register_methods(self) -> Vec<(String, Arc)> { + vec![("/test.Svc/Method".to_string(), Arc::new(self.handler))] + } + } + + let service = SingleMethodService { + handler: OrderHandler { + order: order.clone(), + }, + } + .with_interceptor(per_service); + + let router = RouterBuilder::new() + .chain_interceptor(global) + .add_service(service) + .build(); + + let headers = RequestHeaders::new().with_method_name("/test.Svc/Method"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // Global (1) runs outermost, then per-service (2), then handler (0). + assert_eq!(*order.lock().await, vec![1, 2, 0]); + } +} diff --git a/grpc/src/server/service.rs b/grpc/src/server/service.rs new file mode 100644 index 000000000..5c5cffc13 --- /dev/null +++ b/grpc/src/server/service.rs @@ -0,0 +1,325 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + * + */ + +use std::sync::Arc; + +use crate::server::descriptor::ServiceDescriptor; +use crate::server::interceptor::{HandleExt, Intercept}; +use crate::server::{DynHandle, DynHandleWrapper}; + +/// A gRPC service that can register its methods with a server router. +/// +/// Implementations return their descriptor metadata via [`descriptor()`](Service::descriptor) +/// and produce their method handlers via [`register_methods()`](Service::register_methods). +/// +/// # Example +/// +/// ```ignore +/// use std::sync::Arc; +/// use grpc::server::descriptor::*; +/// use grpc::server::{DynHandle, Service}; +/// +/// struct EchoService { /* ... */ } +/// +/// impl Service for EchoService { +/// fn descriptor(&self) -> ServiceDescriptor { +/// ServiceDescriptor::new( +/// "mypackage.Echo", +/// vec![MethodDescriptor::new("/mypackage.Echo/UnaryEcho", MethodType::Unary)], +/// ) +/// } +/// +/// fn register_methods(self) -> Vec<(String, Arc)> { +/// vec![( +/// "/mypackage.Echo/UnaryEcho".to_string(), +/// Arc::new(self.unary_handler()), +/// )] +/// } +/// } +/// ``` +pub trait Service: Send + 'static { + /// Returns the service descriptor (pure metadata). + /// + /// This provides service and method metadata without registering handlers, + /// enabling use cases like server reflection and service listing. + fn descriptor(&self) -> ServiceDescriptor; + + /// Produces all method handlers for this service as type-erased dynamic handlers + /// paired with their full method path (e.g. `"/mypackage.Echo/UnaryEcho"`). + fn register_methods(self) -> Vec<(String, Arc)>; +} + +/// A service wrapped with an interceptor that applies to all its methods. +/// +/// Created by [`ServiceExt::with_interceptor()`]. The interceptor is applied +/// to each method handler at registration time. +pub struct InterceptedService { + service: S, + interceptor: I, +} + +impl Service for InterceptedService +where + S: Service, + I: Intercept + Clone + Send + Sync + 'static, +{ + fn descriptor(&self) -> ServiceDescriptor { + self.service.descriptor() + } + + fn register_methods(self) -> Vec<(String, Arc)> { + let methods = self.service.register_methods(); + methods + .into_iter() + .map(|(path, handler)| { + let intercepted = + DynHandleWrapper(handler).with_interceptor(self.interceptor.clone()); + (path, Arc::new(intercepted) as Arc) + }) + .collect() + } +} + +/// Extension trait for composing interceptors on services. +pub trait ServiceExt: Service + Sized { + /// Wraps this service with an interceptor that applies to all its methods. + /// + /// This is a pre-registration transformation: the interceptor is applied + /// when the service registers its methods, not at call time. + /// + /// Equivalent to Java's `ServerInterceptors.intercept(service, interceptor)`. + /// + /// # Example + /// + /// ```ignore + /// let rate_limited_greeter = greeter_service.with_interceptor(rate_limiter); + /// ``` + fn with_interceptor(self, interceptor: I) -> InterceptedService { + InterceptedService { + service: self, + interceptor, + } + } +} + +impl ServiceExt for T {} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use tokio::sync::Mutex; + + use super::*; + use crate::client::CallOptions; + use crate::core::RecvMessage; + use crate::server::RequestHeaders; + use crate::server::ResponseStreamItem; + use crate::server::SendOptions; + use crate::server::Trailers; + use crate::server::descriptor::{MethodDescriptor, MethodType, ServiceDescriptor}; + use crate::server::interceptor::Intercept; + use crate::server::router::RouterBuilder; + use crate::server::{Handle, RecvStream, SendStream}; + + struct MockSendStream; + impl SendStream for MockSendStream { + async fn send<'a>( + &mut self, + _item: ResponseStreamItem<'a>, + _options: SendOptions, + ) -> Result<(), ()> { + Ok(()) + } + } + + struct MockRecvStream; + impl RecvStream for MockRecvStream { + async fn next(&mut self, _msg: &mut dyn RecvMessage) -> Option> { + None + } + } + + /// An interceptor that records that it ran by pushing its id. + #[derive(Clone)] + struct TrackingInterceptor { + id: usize, + order: Arc>>, + } + + impl Intercept for TrackingInterceptor { + async fn intercept( + &self, + headers: RequestHeaders, + options: CallOptions, + tx: &mut impl SendStream, + rx: impl RecvStream + 'static, + next: &impl Handle, + ) -> Trailers { + self.order.lock().await.push(self.id); + next.handle(headers, options, tx, rx).await + } + } + + /// A handler that pushes 0 to confirm it was called. + struct TrackingHandler { + order: Arc>>, + } + + impl Handle for TrackingHandler { + async fn handle( + &self, + _headers: RequestHeaders, + _options: CallOptions, + _tx: &mut impl SendStream, + _rx: impl RecvStream + 'static, + ) -> Trailers { + self.order.lock().await.push(0); + Trailers::new(Ok(())) + } + } + + /// A mock service that registers one method with a tracking handler. + struct MockService { + order: Arc>>, + } + + impl Service for MockService { + fn descriptor(&self) -> ServiceDescriptor { + ServiceDescriptor::new( + "test.MockService", + vec![MethodDescriptor::new( + "/test.MockService/Method", + MethodType::Unary, + )], + ) + } + + fn register_methods(self) -> Vec<(String, Arc)> { + vec![( + "/test.MockService/Method".to_string(), + Arc::new(TrackingHandler { order: self.order }), + )] + } + } + + #[test] + fn intercepted_service_preserves_descriptor() { + let order = Arc::new(Mutex::new(Vec::new())); + let svc = MockService { + order: order.clone(), + }; + let interceptor = TrackingInterceptor { + id: 1, + order: order.clone(), + }; + + let intercepted = svc.with_interceptor(interceptor); + let desc = intercepted.descriptor(); + assert_eq!(desc.name(), "test.MockService"); + assert_eq!(desc.methods().len(), 1); + assert_eq!(desc.methods()[0].full_path(), "/test.MockService/Method"); + } + + #[tokio::test] + async fn intercepted_service_applies_interceptor_to_handler() { + let order = Arc::new(Mutex::new(Vec::new())); + let svc = MockService { + order: order.clone(), + }; + let interceptor = TrackingInterceptor { + id: 1, + order: order.clone(), + }; + + // Wrap service with interceptor and register via RouterBuilder. + let intercepted = svc.with_interceptor(interceptor); + let router = RouterBuilder::new().add_service(intercepted).build(); + + // Invoke the handler. + let headers = RequestHeaders::new().with_method_name("/test.MockService/Method"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // Interceptor (1) should run before handler (0). + assert_eq!(*order.lock().await, vec![1, 0]); + } + + #[tokio::test] + async fn intercepted_service_chains_multiple_interceptors() { + let order = Arc::new(Mutex::new(Vec::new())); + let svc = MockService { + order: order.clone(), + }; + let int_a = TrackingInterceptor { + id: 1, + order: order.clone(), + }; + let int_b = TrackingInterceptor { + id: 2, + order: order.clone(), + }; + + // Chain: service -> int_a -> int_b + // Execution: int_b runs first (outermost), then int_a, then handler. + let intercepted = svc.with_interceptor(int_a).with_interceptor(int_b); + let router = RouterBuilder::new().add_service(intercepted).build(); + + let headers = RequestHeaders::new().with_method_name("/test.MockService/Method"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // Outermost interceptor (2) runs first, then (1), then handler (0). + assert_eq!(*order.lock().await, vec![2, 1, 0]); + } + + #[tokio::test] + async fn service_without_interceptor_handler_runs_directly() { + let order = Arc::new(Mutex::new(Vec::new())); + let svc = MockService { + order: order.clone(), + }; + + let router = RouterBuilder::new().add_service(svc).build(); + + let headers = RequestHeaders::new().with_method_name("/test.MockService/Method"); + let mut tx = MockSendStream; + let rx = MockRecvStream; + let trailers = router + .handle(headers, CallOptions::default(), &mut tx, rx) + .await; + + assert!(trailers.status().is_ok()); + // Only handler (0), no interceptors. + assert_eq!(*order.lock().await, vec![0]); + } +}