Add server interceptor acting as a middleware (#762)

* Add server interceptor acting as a middleware

* Remove unused import

* Add changelog entry

* Inline variable

* Bump version

* Fix infinite recursion

Author: ThHareau

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 4.1.0
+* Add a `serverInterceptors` argument to `ConnectionServer`. These interceptors are acting 
+  as middleware, wrapping a `ServiceMethod` invocation.
+
 ## 4.0.4
 
 * Allow the latest `package:googleapis_auth`.

lib/grpc.dart

@@ -42,7 +42,8 @@ export 'src/client/proxy.dart' show Proxy;
 export 'src/client/transport/http2_credentials.dart'
     show BadCertificateHandler, allowBadCertificates, ChannelCredentials;
 export 'src/server/call.dart' show ServiceCall;
-export 'src/server/interceptor.dart' show Interceptor;
+export 'src/server/interceptor.dart'
+    show Interceptor, ServerInterceptor, ServerStreamingInvoker;
 export 'src/server/server.dart'
     show
         ServerCredentials,

lib/src/server/handler.dart

@@ -37,6 +37,7 @@ class ServerHandler extends ServiceCall {
   final ServerTransportStream _stream;
   final ServiceLookup _serviceLookup;
   final List<Interceptor> _interceptors;
+  final List<ServerInterceptor> _serverInterceptors;
   final CodecRegistry? _codecRegistry;
   final GrpcErrorHandler? _errorHandler;
 
@@ -83,6 +84,7 @@ class ServerHandler extends ServiceCall {
     required ServerTransportStream stream,
     required ServiceLookup serviceLookup,
     required List<Interceptor> interceptors,
+    required List<ServerInterceptor> serverInterceptors,
     required CodecRegistry? codecRegistry,
     X509Certificate? clientCertificate,
     InternetAddress? remoteAddress,
@@ -94,7 +96,8 @@ class ServerHandler extends ServiceCall {
         _codecRegistry = codecRegistry,
         _clientCertificate = clientCertificate,
         _remoteAddress = remoteAddress,
-        _errorHandler = errorHandler;
+        _errorHandler = errorHandler,
+        _serverInterceptors = serverInterceptors;
 
   @override
   DateTime? get deadline => _deadline;
@@ -239,7 +242,7 @@ class ServerHandler extends ServiceCall {
       return;
     }
 
-    _responses = _descriptor.handle(this, requests.stream);
+    _responses = _descriptor.handle(this, requests.stream, _serverInterceptors);
 
     _responseSubscription = _responses.listen(_onResponse,
         onError: _onResponseError,

lib/src/server/interceptor.dart

@@ -27,3 +27,18 @@ import 'service.dart';
 /// If the interceptor returns null, the corresponding [ServiceMethod] of [Service] will be called.
 typedef Interceptor = FutureOr<GrpcError?> Function(
     ServiceCall call, ServiceMethod method);
+
+typedef ServerStreamingInvoker<Q, R> = Stream<R> Function(
+    ServiceCall call, ServiceMethod<Q, R> method, Stream<Q> requests);
+
+/// A gRPC Interceptor.
+///
+/// An interceptor is called around the corresponding [ServiceMethod] invocation.
+/// If the interceptor throws [GrpcError], the error will be returned as a response. [ServiceMethod] wouldn't be called if the error is thrown before calling the invoker.
+/// If the interceptor modifies the provided stream, the invocation will continue with the provided stream.
+abstract class ServerInterceptor {
+  Stream<R> intercept<Q, R>(ServiceCall call, ServiceMethod<Q, R> method,
+      Stream<Q> requests, ServerStreamingInvoker<Q, R> invoker) {
+    return invoker(call, method, requests);
+  }
+}

lib/src/server/server.dart

@@ -87,6 +87,7 @@ class ServerTlsCredentials extends ServerCredentials {
 class ConnectionServer {
   final Map<String, Service> _services = {};
   final List<Interceptor> _interceptors;
+  final List<ServerInterceptor> _serverInterceptors;
   final CodecRegistry? _codecRegistry;
   final GrpcErrorHandler? _errorHandler;
   final ServerKeepAliveOptions _keepAliveOptions;
@@ -100,11 +101,13 @@ class ConnectionServer {
   ConnectionServer(
     List<Service> services, [
     List<Interceptor> interceptors = const <Interceptor>[],
+    List<ServerInterceptor> serverInterceptors = const <ServerInterceptor>[],
     CodecRegistry? codecRegistry,
     GrpcErrorHandler? errorHandler,
     this._keepAliveOptions = const ServerKeepAliveOptions(),
   ])  : _codecRegistry = codecRegistry,
         _interceptors = interceptors,
+        _serverInterceptors = serverInterceptors,
         _errorHandler = errorHandler {
     for (final service in services) {
       _services[service.$name] = service;
@@ -168,6 +171,7 @@ class ConnectionServer {
         stream: stream,
         serviceLookup: lookupService,
         interceptors: _interceptors,
+        serverInterceptors: _serverInterceptors,
         codecRegistry: _codecRegistry,
         // ignore: unnecessary_cast
         clientCertificate: clientCertificate as io_bits.X509Certificate?,
@@ -201,11 +205,13 @@ class Server extends ConnectionServer {
     required List<Service> services,
     ServerKeepAliveOptions keepAliveOptions = const ServerKeepAliveOptions(),
     List<Interceptor> interceptors = const <Interceptor>[],
+    List<ServerInterceptor> serverInterceptors = const <ServerInterceptor>[],
     CodecRegistry? codecRegistry,
     GrpcErrorHandler? errorHandler,
   }) : super(
           services,
           interceptors,
+          serverInterceptors,
           codecRegistry,
           errorHandler,
           keepAliveOptions,
@@ -308,6 +314,7 @@ class Server extends ConnectionServer {
       stream: stream,
       serviceLookup: lookupService,
       interceptors: _interceptors,
+      serverInterceptors: _serverInterceptors,
       codecRegistry: _codecRegistry,
       // ignore: unnecessary_cast
       clientCertificate: clientCertificate as io_bits.X509Certificate?,

lib/src/server/service.dart

@@ -17,6 +17,7 @@ import 'dart:async';
 
 import '../shared/status.dart';
 import 'call.dart';
+import 'interceptor.dart';
 
 /// Definition of a gRPC service method.
 class ServiceMethod<Q, R> {
@@ -48,19 +49,42 @@ class ServiceMethod<Q, R> {
 
   List<int> serialize(dynamic response) => responseSerializer(response as R);
 
-  Stream<R> handle(ServiceCall call, Stream<Q> requests) {
-    if (streamingResponse) {
-      if (streamingRequest) {
-        return handler(call, requests);
-      } else {
-        return handler(call, _toSingleFuture(requests));
-      }
-    } else {
-      final response = streamingRequest
-          ? handler(call, requests)
-          : handler(call, _toSingleFuture(requests));
-      return response.asStream();
+  ServerStreamingInvoker<Q, R> _createCall() => ((
+        ServiceCall call,
+        ServiceMethod<Q, R> method,
+        Stream<Q> requests,
+      ) {
+        if (streamingResponse) {
+          if (streamingRequest) {
+            return handler(call, requests);
+          } else {
+            return handler(call, _toSingleFuture(requests));
+          }
+        } else {
+          final response = streamingRequest
+              ? handler(call, requests)
+              : handler(call, _toSingleFuture(requests));
+          return response.asStream();
+        }
+      });
+
+  Stream<R> handle(
+    ServiceCall call,
+    Stream<Q> requests,
+    List<ServerInterceptor> interceptors,
+  ) {
+    var invoker = _createCall();
+
+    for (final interceptor in interceptors.reversed) {
+      final delegate = invoker;
+      // invoker is actually reassigned in the same scope as the above function,
+      // reassigning invoker in delegate is required to avoid an infinite
+      // recursion
+      invoker = (call, method, requests) =>
+          interceptor.intercept<Q, R>(call, method, requests, delegate);
     }
+
+    return invoker(call, this, requests);
   }
 
   Future<Q> _toSingleFuture(Stream<Q> stream) {

pubspec.yaml

@@ -1,5 +1,5 @@
 name: grpc
-version: 4.0.4
+version: 4.1.0
 description: Dart implementation of gRPC, a high performance, open-source universal RPC framework.
 repository: https://github.com/grpc/grpc-dart