Don't push ServerInterceptorTargets down into the RPC executor

Author: gjcairo

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

@@ -34,7 +34,7 @@ struct ServerRPCExecutor {
     >,
     deserializer: some MessageDeserializer<Input>,
     serializer: some MessageSerializer<Output>,
-    interceptors: [ServerInterceptorTarget],
+    interceptors: [any ServerInterceptor],
     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: [ServerInterceptorTarget],
+    interceptors: [any ServerInterceptor],
     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: [ServerInterceptorTarget],
+    interceptors: [any ServerInterceptor],
     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: [ServerInterceptorTarget],
+    interceptors: [any ServerInterceptor],
     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: [ServerInterceptorTarget],
+    interceptors: [any ServerInterceptor],
     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<ServerInterceptorTarget>.Iterator,
+    iterator: Array<any ServerInterceptor>.Iterator,
     finally: @escaping @Sendable (
       _ request: StreamingServerRequest<Input>,
       _ context: ServerContext
@@ -313,29 +313,20 @@ extension ServerRPCExecutor {
     var iterator = iterator
 
     switch iterator.next() {
-    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)
-        }
-      } else {
-        return try await self._intercept(
+    case .some(let interceptor):
+      let iter = iterator
+      do {
+        return try await interceptor.intercept(
           request: request,
-          context: context,
-          iterator: iterator,
-          finally: finally
-        )
+          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)
       }
 
     case .none:

Sources/GRPCCore/Call/Server/RPCRouter.swift

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

Sources/GRPCCore/GRPCServer.swift

@@ -173,7 +173,7 @@ public final class GRPCServer: Sendable {
   public convenience init(
     transport: any ServerTransport,
     services: [any RegistrableRPCService],
-    interceptors: [ServerInterceptorTarget] = []
+    interceptors: [ServerInterceptorTarget]
   ) {
     var router = RPCRouter()
     for service in services {
@@ -218,7 +218,7 @@ public final class GRPCServer: Sendable {
   public init(
     transport: any ServerTransport,
     router: RPCRouter,
-    interceptors: [ServerInterceptorTarget] = []
+    interceptors: [ServerInterceptorTarget]
   ) {
     self.state = Mutex(.notStarted)
     self.transport = transport
@@ -249,7 +249,13 @@ public final class GRPCServer: Sendable {
 
     do {
       try await transport.listen { stream, context in
-        await self.router.handle(stream: stream, context: context, interceptors: self.interceptors)
+        await self.router.handle(
+          stream: stream,
+          context: context,
+          interceptors: self.interceptors
+            .filter { $0.applies(to: context.descriptor) }
+            .map { $0.interceptor }
+        )
       }
     } catch {
       throw RuntimeError(

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

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

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

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