Add ServerInterceptorTarget

Author: gjcairo

Sources/GRPCCore/Call/Server/Internal/ServerRPCExecutor.swift

@@ -34,7 +34,7 @@ struct ServerRPCExecutor {
     >,
     deserializer: some MessageDeserializer<Input>,
     serializer: some MessageSerializer<Output>,
-    interceptors: [any ServerInterceptor],
+    interceptors: [ServerInterceptorTarget],
     handler: @Sendable @escaping (
       _ request: StreamingServerRequest<Input>,
       _ context: ServerContext
@@ -72,7 +72,7 @@ struct ServerRPCExecutor {
     outbound: RPCWriter<RPCResponsePart>.Closable,
     deserializer: some MessageDeserializer<Input>,
     serializer: some MessageSerializer<Output>,
-    interceptors: [any ServerInterceptor],
+    interceptors: [ServerInterceptorTarget],
     handler: @escaping @Sendable (
       _ request: StreamingServerRequest<Input>,
       _ context: ServerContext
@@ -113,7 +113,7 @@ struct ServerRPCExecutor {
     outbound: RPCWriter<RPCResponsePart>.Closable,
     deserializer: some MessageDeserializer<Input>,
     serializer: some MessageSerializer<Output>,
-    interceptors: [any ServerInterceptor],
+    interceptors: [ServerInterceptorTarget],
     handler: @escaping @Sendable (
       _ request: StreamingServerRequest<Input>,
       _ context: ServerContext
@@ -153,7 +153,7 @@ struct ServerRPCExecutor {
     outbound: RPCWriter<RPCResponsePart>.Closable,
     deserializer: some MessageDeserializer<Input>,
     serializer: some MessageSerializer<Output>,
-    interceptors: [any ServerInterceptor],
+    interceptors: [ServerInterceptorTarget],
     handler: @escaping @Sendable (
       _ request: StreamingServerRequest<Input>,
       _ context: ServerContext
@@ -286,7 +286,7 @@ extension ServerRPCExecutor {
   static func _intercept<Input, Output>(
     request: StreamingServerRequest<Input>,
     context: ServerContext,
-    interceptors: [any ServerInterceptor],
+    interceptors: [ServerInterceptorTarget],
     finally: @escaping @Sendable (
       _ request: StreamingServerRequest<Input>,
       _ context: ServerContext
@@ -304,7 +304,7 @@ extension ServerRPCExecutor {
   static func _intercept<Input, Output>(
     request: StreamingServerRequest<Input>,
     context: ServerContext,
-    iterator: Array<any ServerInterceptor>.Iterator,
+    iterator: Array<ServerInterceptorTarget>.Iterator,
     finally: @escaping @Sendable (
       _ request: StreamingServerRequest<Input>,
       _ context: ServerContext
@@ -313,17 +313,29 @@ extension ServerRPCExecutor {
     var iterator = iterator
 
     switch iterator.next() {
-    case .some(let interceptor):
-      let iter = iterator
-      do {
-        return try await interceptor.intercept(request: request, context: context) {
-          try await self._intercept(request: $0, context: $1, iterator: iter, finally: finally)
+    case .some(let interceptorTarget):
+      if interceptorTarget.applies(to: context.descriptor) {
+        let iter = iterator
+        do {
+          return try await interceptorTarget.interceptor.intercept(
+            request: request,
+            context: context
+          ) {
+            try await self._intercept(request: $0, context: $1, iterator: iter, finally: finally)
+          }
+        } catch let error as RPCError {
+          return StreamingServerResponse(error: error)
+        } catch let other {
+          let error = RPCError(code: .unknown, message: "", cause: other)
+          return StreamingServerResponse(error: error)
         }
-      } catch let error as RPCError {
-        return StreamingServerResponse(error: error)
-      } catch let other {
-        let error = RPCError(code: .unknown, message: "", cause: other)
-        return StreamingServerResponse(error: error)
+      } else {
+        return try await self._intercept(
+          request: request,
+          context: context,
+          iterator: iterator,
+          finally: finally
+        )
       }
 
     case .none:

Sources/GRPCCore/Call/Server/RPCRouter.swift

@@ -43,7 +43,7 @@ public struct RPCRouter: Sendable {
           RPCWriter<RPCResponsePart>.Closable
         >,
         _ context: ServerContext,
-        _ interceptors: [any ServerInterceptor]
+        _ interceptors: [ServerInterceptorTarget]
       ) async -> Void
 
     @inlinable
@@ -75,7 +75,7 @@ public struct RPCRouter: Sendable {
         RPCWriter<RPCResponsePart>.Closable
       >,
       context: ServerContext,
-      interceptors: [any ServerInterceptor]
+      interceptors: [ServerInterceptorTarget]
     ) async {
       await self._fn(stream, context, interceptors)
     }
@@ -151,7 +151,7 @@ extension RPCRouter {
       RPCWriter<RPCResponsePart>.Closable
     >,
     context: ServerContext,
-    interceptors: [any ServerInterceptor]
+    interceptors: [ServerInterceptorTarget]
   ) async {
     if let handler = self.handlers[stream.descriptor] {
       await handler.handle(stream: stream, context: context, interceptors: interceptors)

Sources/GRPCCore/Call/Server/ServerInterceptor.swift

@@ -21,10 +21,11 @@
 /// been returned from a service. They are typically used for cross-cutting concerns like filtering
 /// requests, validating messages, logging additional data, and tracing.
 ///
-/// Interceptors are registered with the server apply to all RPCs. If you need to modify the
-/// behavior of an interceptor on a per-RPC basis then you can use the
-/// ``ServerContext/descriptor`` to determine which RPC is being called and
-/// conditionalise behavior accordingly.
+/// Interceptors are registered with the server via ``ServerInterceptorTarget``s.
+/// You may register them for all services registered with a server, for RPCs directed to specific services, or
+/// for RPCs directed to specific methods. If you need to modify the behavior of an interceptor on a
+/// per-RPC basis in more detail, then you can use the ``ServerContext/descriptor`` to determine
+/// which RPC is being called and conditionalise behavior accordingly.
 ///
 /// ## RPC filtering
 ///
@@ -33,19 +34,19 @@
 /// demonstrates this.
 ///
 /// ```swift
-/// struct AuthServerInterceptor: Sendable {
+/// struct AuthServerInterceptor: ServerInterceptor {
 ///   let isAuthorized: @Sendable (String, MethodDescriptor) async throws -> Void
 ///
 ///   func intercept<Input: Sendable, Output: Sendable>(
 ///     request: StreamingServerRequest<Input>,
-///     context: ServerInterceptorContext,
+///     context: ServerContext,
 ///     next: @Sendable (
 ///       _ request: StreamingServerRequest<Input>,
-///       _ context: ServerInterceptorContext
+///       _ context: ServerContext
 ///     ) async throws -> StreamingServerResponse<Output>
 ///   ) async throws -> StreamingServerResponse<Output> {
 ///     // Extract the auth token.
-///     guard let token = request.metadata["authorization"] else {
+///     guard let token = request.metadata[stringValues: "authorization"].first(where: { _ in true }) else {
 ///       throw RPCError(code: .unauthenticated, message: "Not authenticated")
 ///     }
 ///

Sources/GRPCCore/Call/Server/ServerInterceptorTarget.swift

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2024, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/// A `ServerInterceptorTarget` describes to which RPCs a server interceptor should be applied.
+///
+/// You can configure a server interceptor to be applied to:
+/// - all RPCs and services;
+/// - requests directed only to specific services registered with your server; or
+/// - requests directed only to specific methods (of a specific service).
+///
+/// - SeeAlso: ``ServerInterceptor`` for more information on server interceptors, and
+///  ``ClientInterceptorTarget`` for the client-side version of this type.
+public struct ServerInterceptorTarget: Sendable {
+  internal enum Wrapped: Sendable {
+    case allServices(interceptor: any ServerInterceptor)
+    case serviceSpecific(interceptor: any ServerInterceptor, services: [String])
+    case methodSpecific(interceptor: any ServerInterceptor, methods: [MethodDescriptor])
+  }
+
+  /// A target specifying an interceptor that applies to all RPCs across all services registered with this server.
+  /// - Parameter interceptor: The interceptor to register with the server.
+  /// - Returns: A ``ServerInterceptorTarget``.
+  public static func allServices(
+    interceptor: any ServerInterceptor
+  ) -> Self {
+    Self(wrapped: .allServices(interceptor: interceptor))
+  }
+
+  /// A target specifying an interceptor that applies to RPCs directed only to the specified services.
+  /// - Parameters:
+  ///   - interceptor: The interceptor to register with the server.
+  ///   - services: The list of service names for which this interceptor should intercept RPCs.
+  /// - Returns: A ``ServerInterceptorTarget``.
+  public static func serviceSpecific(
+    interceptor: any ServerInterceptor,
+    services: [String]
+  ) -> Self {
+    Self(
+      wrapped: .serviceSpecific(
+        interceptor: interceptor,
+        services: services
+      )
+    )
+  }
+
+  /// A target specifying an interceptor that applies to RPCs directed only to the specified service methods.
+  /// - Parameters:
+  ///   - interceptor: The interceptor to register with the server.
+  ///   - services: The list of method descriptors for which this interceptor should intercept RPCs.
+  /// - Returns: A ``ServerInterceptorTarget``.
+  public static func methodSpecific(
+    interceptor: any ServerInterceptor,
+    methods: [MethodDescriptor]
+  ) -> Self {
+    Self(
+      wrapped: .methodSpecific(
+        interceptor: interceptor,
+        methods: methods
+      )
+    )
+  }
+
+  private let wrapped: Wrapped
+
+  private init(wrapped: Wrapped) {
+    self.wrapped = wrapped
+  }
+
+  /// Get the ``ServerInterceptor`` associated with this ``ServerInterceptorTarget``.
+  public var interceptor: any ServerInterceptor {
+    switch self.wrapped {
+    case .allServices(let interceptor):
+      return interceptor
+    case .serviceSpecific(let interceptor, _):
+      return interceptor
+    case .methodSpecific(let interceptor, _):
+      return interceptor
+    }
+  }
+
+  /// Returns whether this ``ServerInterceptorTarget`` applies to the given `descriptor`.
+  /// - Parameter descriptor: A ``MethodDescriptor`` for which to test whether this interceptor applies.
+  /// - Returns: `true` if this interceptor applies to the given `descriptor`, or `false` otherwise.
+  public func applies(to descriptor: MethodDescriptor) -> Bool {
+    switch self.wrapped {
+    case .allServices:
+      return true
+    case .serviceSpecific(_, let services):
+      return services.contains(descriptor.service)
+    case .methodSpecific(_, let methods):
+      return methods.contains(descriptor)
+    }
+  }
+}

Sources/GRPCCore/GRPCServer.swift

@@ -48,7 +48,7 @@ private import Synchronization
 /// let server = GRPCServer(
 ///   transport: inProcessTransport.server,
 ///   services: [greeter, echo],
-///   interceptors: [statsRecorder]
+///   interceptors: [.allServices(interceptor: statsRecorder)]
 /// )
 /// ```
 ///
@@ -78,13 +78,12 @@ public final class GRPCServer: Sendable {
   /// The services registered which the server is serving.
   private let router: RPCRouter
 
-  /// A collection of ``ServerInterceptor`` implementations which are applied to all accepted
-  /// RPCs.
+  /// A collection of ``ServerInterceptorTarget``s which may be applied to all accepted RPCs.
   ///
   /// RPCs are intercepted in the order that interceptors are added. That is, a request received
   /// from the client will first be intercepted by the first added interceptor followed by the
   /// second, and so on.
-  private let interceptors: [any ServerInterceptor]
+  private let interceptors: [ServerInterceptorTarget]
 
   /// The state of the server.
   private let state: Mutex<State>
@@ -152,7 +151,7 @@ public final class GRPCServer: Sendable {
   public convenience init(
     transport: any ServerTransport,
     services: [any RegistrableRPCService],
-    interceptors: [any ServerInterceptor] = []
+    interceptors: [ServerInterceptorTarget] = []
   ) {
     var router = RPCRouter()
     for service in services {
@@ -175,7 +174,7 @@ public final class GRPCServer: Sendable {
   public init(
     transport: any ServerTransport,
     router: RPCRouter,
-    interceptors: [any ServerInterceptor] = []
+    interceptors: [ServerInterceptorTarget] = []
   ) {
     self.state = Mutex(.notStarted)
     self.transport = transport

Tests/GRPCCoreTests/Call/Server/Internal/ServerRPCExecutorTestSupport/ServerRPCExecutorTestHarness.swift

@@ -46,9 +46,9 @@ struct ServerRPCExecutorTestHarness {
     }
   }
 
-  let interceptors: [any ServerInterceptor]
+  let interceptors: [ServerInterceptorTarget]
 
-  init(interceptors: [any ServerInterceptor] = []) {
+  init(interceptors: [ServerInterceptorTarget] = []) {
     self.interceptors = interceptors
   }
 

Tests/GRPCCoreTests/Call/Server/Internal/ServerRPCExecutorTests.swift

@@ -258,7 +258,9 @@ final class ServerRPCExecutorTests: XCTestCase {
     )
 
     // The interceptor skips the handler altogether.
-    let harness = ServerRPCExecutorTestHarness(interceptors: [.rejectAll(with: error)])
+    let harness = ServerRPCExecutorTestHarness(interceptors: [
+      .allServices(interceptor: .rejectAll(with: error))
+    ])
     try await harness.execute(
       deserializer: IdentityDeserializer(),
       serializer: IdentitySerializer()
@@ -288,8 +290,8 @@ final class ServerRPCExecutorTests: XCTestCase {
     // The interceptor skips the handler altogether.
     let harness = ServerRPCExecutorTestHarness(
       interceptors: [
-        .requestCounter(counter1),
-        .requestCounter(counter2),
+        .allServices(interceptor: .requestCounter(counter1)),
+        .allServices(interceptor: .requestCounter(counter2)),
       ]
     )
 
@@ -312,9 +314,9 @@ final class ServerRPCExecutorTests: XCTestCase {
     // The interceptor skips the handler altogether.
     let harness = ServerRPCExecutorTestHarness(
       interceptors: [
-        .requestCounter(counter1),
-        .rejectAll(with: RPCError(code: .unavailable, message: "")),
-        .requestCounter(counter2),
+        .allServices(interceptor: .requestCounter(counter1)),
+        .allServices(interceptor: .rejectAll(with: RPCError(code: .unavailable, message: ""))),
+        .allServices(interceptor: .requestCounter(counter2)),
       ]
     )
 
@@ -333,7 +335,9 @@ final class ServerRPCExecutorTests: XCTestCase {
 
   func testThrowingInterceptor() async throws {
     let harness = ServerRPCExecutorTestHarness(
-      interceptors: [.throwError(RPCError(code: .unavailable, message: "Unavailable"))]
+      interceptors: [
+        .allServices(interceptor: .throwError(RPCError(code: .unavailable, message: "Unavailable")))
+      ]
     )
 
     try await harness.execute(handler: .echo) { inbound in