Call to onMetadata handles exceptions (#342)

Co-authored-by: Vyacheslav Egorov <vegorov@google.com>

lib/src/server/handler.dart

@@ -131,6 +131,18 @@ class ServerHandler_ extends ServiceCall {
     _startStreamingRequest();
   }
 
+  GrpcError _onMetadata() {
+    try {
+      _service.$onMetadata(this);
+    } on GrpcError catch (error) {
+      return error;
+    } catch (error) {
+      final grpcError = GrpcError.internal(error.toString());
+      return grpcError;
+    }
+    return null;
+  }
+
   Future<GrpcError> _applyInterceptors() async {
     try {
       for (final interceptor in _interceptors) {
@@ -150,7 +162,19 @@ class ServerHandler_ extends ServiceCall {
     _requests = _descriptor.createRequestStream(_incomingSubscription);
     _incomingSubscription.onData(_onDataActive);
 
-    _service.$onMetadata(this);
+    final error = _onMetadata();
+    if (error != null) {
+      if (!_requests.isClosed) {
+        _requests
+          ..addError(error)
+          ..close();
+      }
+      _sendError(error);
+      _onDone();
+      _stream.terminate();
+      return;
+    }
+
     _responses = _descriptor.handle(this, _requests.stream);
 
     _responseSubscription = _responses.listen(_onResponse,

test/round_trip_test.dart

@@ -35,6 +35,12 @@ class TestService extends Service {
   }
 }
 
+class TestServiceWithOnMetadataException extends TestService {
+  void $onMetadata(ServiceCall context) {
+    throw Exception('business exception');
+  }
+}
+
 class FixedConnectionClientChannel extends ClientChannelBase {
   final Http2ClientConnection clientConnection;
   List<ConnectionState> states = <ConnectionState>[];
@@ -81,4 +87,18 @@ main() async {
     expect(await testClient.stream(1).toList(), [1, 2, 3]);
     server.shutdown();
   });
+
+  test('exception in onMetadataException', () async {
+    final Server server = Server([TestServiceWithOnMetadataException()]);
+    await server.serve(address: 'localhost', port: 0);
+
+    final channel = FixedConnectionClientChannel(Http2ClientConnection(
+      'localhost',
+      server.port,
+      ChannelOptions(credentials: ChannelCredentials.insecure()),
+    ));
+    final testClient = TestClient(channel);
+    await expectLater(testClient.stream(1).toList(), throwsA(isA<GrpcError>()));
+    await server.shutdown();
+  });
 }