Basic server interoperability tests. (#25)

Implements the basic server interoperability tests for gRPC compliance.
Compressed tests, TLS, and authentication are not implemented yet, but
basic test cases pass against the C++ gRPC interoperability client.

.travis.yml

@@ -22,6 +22,7 @@ matrix:
     script:
     - dartanalyzer lib test
     - for example in example/*; do (cd $example; echo [Analyzing $example]; pub get; dartanalyzer .); done
+    - (cd interop; echo [Analyzing interop]; pub get; dartanalyzer .)
 
 
 # Only building master means that we don't run two builds for each pull request.
@@ -31,7 +32,7 @@ branches:
 os:
   - linux
   - osx
-  
+
 cache:
   directories:
     - $HOME/.pub-cache

interop/bin/server.dart

@@ -0,0 +1,113 @@
+// Copyright (c) 2017, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:args/args.dart';
+import 'package:grpc/grpc.dart';
+
+import 'package:interop/src/generated/empty.pb.dart';
+import 'package:interop/src/generated/messages.pb.dart';
+import 'package:interop/src/generated/test.pbgrpc.dart';
+
+const _headerEchoKey = 'x-grpc-test-echo-initial';
+const _trailerEchoKey = 'x-grpc-test-echo-trailing-bin';
+
+class TestService extends TestServiceBase {
+  @override
+  void $onMetadata(ServiceCall context) {
+    final headerEcho = context.clientMetadata[_headerEchoKey];
+    if (headerEcho != null) {
+      context.headers[_headerEchoKey] = headerEcho;
+    }
+    final trailerEcho = context.clientMetadata[_trailerEchoKey];
+    if (trailerEcho != null) {
+      context.trailers[_trailerEchoKey] = trailerEcho;
+    }
+  }
+
+  @override
+  Future<Empty> emptyCall(ServiceCall call, Empty request) async {
+    return new Empty();
+  }
+
+  @override
+  Future<SimpleResponse> unaryCall(
+      ServiceCall call, SimpleRequest request) async {
+    if (request.responseStatus.code != 0) {
+      throw new GrpcError.custom(
+          request.responseStatus.code, request.responseStatus.message);
+    }
+    final payload = new Payload()
+      ..body = new List.filled(request.responseSize, 0);
+    return new SimpleResponse()..payload = payload;
+  }
+
+  @override
+  Future<SimpleResponse> cacheableUnaryCall(
+      ServiceCall call, SimpleRequest request) async {
+    final timestamp = new DateTime.now().microsecond * 1000;
+    final responsePayload = new Payload()..body = ASCII.encode('$timestamp');
+    return new SimpleResponse()..payload = responsePayload;
+  }
+
+  @override
+  Future<StreamingInputCallResponse> streamingInputCall(
+      ServiceCall call, Stream<StreamingInputCallRequest> request) async {
+    final aggregatedPayloadSize = await request.fold(
+        0, (size, message) => size + message.payload.body.length);
+    return new StreamingInputCallResponse()
+      ..aggregatedPayloadSize = aggregatedPayloadSize;
+  }
+
+  Payload _payloadForRequest(ResponseParameters entry) =>
+      new Payload()..body = new List.filled(entry.size, 0);
+
+  @override
+  Stream<StreamingOutputCallResponse> streamingOutputCall(
+      ServiceCall call, StreamingOutputCallRequest request) async* {
+    for (final entry in request.responseParameters) {
+      if (entry.intervalUs > 0) {
+        await new Future.delayed(new Duration(microseconds: entry.intervalUs));
+      }
+      yield new StreamingOutputCallResponse()
+        ..payload = _payloadForRequest(entry);
+    }
+  }
+
+  StreamingOutputCallResponse _responseForRequest(
+      StreamingOutputCallRequest request) {
+    if (request.responseStatus.code != 0) {
+      throw new GrpcError.custom(
+          request.responseStatus.code, request.responseStatus.message);
+    }
+    return new StreamingOutputCallResponse()
+      ..payload = _payloadForRequest(request.responseParameters[0]);
+  }
+
+  @override
+  Stream<StreamingOutputCallResponse> fullDuplexCall(
+      ServiceCall call, Stream<StreamingOutputCallRequest> request) async* {
+    yield* request.map(_responseForRequest);
+  }
+
+  @override
+  Stream<StreamingOutputCallResponse> halfDuplexCall(
+      ServiceCall call, Stream<StreamingOutputCallRequest> request) async* {
+    final bufferedResponses = await request.map(_responseForRequest).toList();
+    yield* new Stream.fromIterable(bufferedResponses);
+  }
+}
+
+Future<Null> main(List<String> args) async {
+  final argumentParser = new ArgParser();
+  argumentParser.addOption('port', defaultsTo: '8080');
+  argumentParser.addOption('use_tls', defaultsTo: 'false');
+  final arguments = argumentParser.parse(args);
+  final port = int.parse(arguments['port']);
+  final server = new Server(port: port)..addService(new TestService());
+  await server.serve();
+  print('Server listening...');
+}

interop/lib/src/generated/empty.pb.dart

@@ -0,0 +1,37 @@
+///
+//  Generated code. Do not modify.
+///
+// ignore_for_file: non_constant_identifier_names
+// ignore_for_file: library_prefixes
+library grpc.testing_empty;
+
+// ignore: UNUSED_SHOWN_NAME
+import 'dart:core' show int, bool, double, String, List, override;
+
+import 'package:protobuf/protobuf.dart';
+
+class Empty extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('Empty')
+    ..hasRequiredFields = false;
+
+  Empty() : super();
+  Empty.fromBuffer(List<int> i, [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  Empty.fromJson(String i, [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  Empty clone() => new Empty()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static Empty create() => new Empty();
+  static PbList<Empty> createRepeated() => new PbList<Empty>();
+  static Empty getDefault() {
+    if (_defaultInstance == null) _defaultInstance = new _ReadonlyEmpty();
+    return _defaultInstance;
+  }
+
+  static Empty _defaultInstance;
+  static void $checkItem(Empty v) {
+    if (v is! Empty) checkItemFailed(v, 'Empty');
+  }
+}
+
+class _ReadonlyEmpty extends Empty with ReadonlyMessageMixin {}

interop/lib/src/generated/messages.pb.dart

@@ -0,0 +1,656 @@
+///
+//  Generated code. Do not modify.
+///
+// ignore_for_file: non_constant_identifier_names
+// ignore_for_file: library_prefixes
+library grpc.testing_messages;
+
+// ignore: UNUSED_SHOWN_NAME
+import 'dart:core' show int, bool, double, String, List, override;
+
+import 'package:protobuf/protobuf.dart';
+
+import 'messages.pbenum.dart';
+
+export 'messages.pbenum.dart';
+
+class BoolValue extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('BoolValue')
+    ..a/*<bool>*/(1, 'value', PbFieldType.OB)
+    ..hasRequiredFields = false;
+
+  BoolValue() : super();
+  BoolValue.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  BoolValue.fromJson(String i, [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  BoolValue clone() => new BoolValue()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static BoolValue create() => new BoolValue();
+  static PbList<BoolValue> createRepeated() => new PbList<BoolValue>();
+  static BoolValue getDefault() {
+    if (_defaultInstance == null) _defaultInstance = new _ReadonlyBoolValue();
+    return _defaultInstance;
+  }
+
+  static BoolValue _defaultInstance;
+  static void $checkItem(BoolValue v) {
+    if (v is! BoolValue) checkItemFailed(v, 'BoolValue');
+  }
+
+  bool get value => $_get(0, 1, false);
+  set value(bool v) {
+    $_setBool(0, 1, v);
+  }
+
+  bool hasValue() => $_has(0, 1);
+  void clearValue() => clearField(1);
+}
+
+class _ReadonlyBoolValue extends BoolValue with ReadonlyMessageMixin {}
+
+class Payload extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('Payload')
+    ..e/*<PayloadType>*/(1, 'type', PbFieldType.OE, PayloadType.COMPRESSABLE,
+        PayloadType.valueOf)
+    ..a/*<List<int>>*/(2, 'body', PbFieldType.OY)
+    ..hasRequiredFields = false;
+
+  Payload() : super();
+  Payload.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  Payload.fromJson(String i, [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  Payload clone() => new Payload()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static Payload create() => new Payload();
+  static PbList<Payload> createRepeated() => new PbList<Payload>();
+  static Payload getDefault() {
+    if (_defaultInstance == null) _defaultInstance = new _ReadonlyPayload();
+    return _defaultInstance;
+  }
+
+  static Payload _defaultInstance;
+  static void $checkItem(Payload v) {
+    if (v is! Payload) checkItemFailed(v, 'Payload');
+  }
+
+  PayloadType get type => $_get(0, 1, null);
+  set type(PayloadType v) {
+    setField(1, v);
+  }
+
+  bool hasType() => $_has(0, 1);
+  void clearType() => clearField(1);
+
+  List<int> get body => $_get(1, 2, null);
+  set body(List<int> v) {
+    $_setBytes(1, 2, v);
+  }
+
+  bool hasBody() => $_has(1, 2);
+  void clearBody() => clearField(2);
+}
+
+class _ReadonlyPayload extends Payload with ReadonlyMessageMixin {}
+
+class EchoStatus extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('EchoStatus')
+    ..a/*<int>*/(1, 'code', PbFieldType.O3)
+    ..a/*<String>*/(2, 'message', PbFieldType.OS)
+    ..hasRequiredFields = false;
+
+  EchoStatus() : super();
+  EchoStatus.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  EchoStatus.fromJson(String i, [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  EchoStatus clone() => new EchoStatus()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static EchoStatus create() => new EchoStatus();
+  static PbList<EchoStatus> createRepeated() => new PbList<EchoStatus>();
+  static EchoStatus getDefault() {
+    if (_defaultInstance == null) _defaultInstance = new _ReadonlyEchoStatus();
+    return _defaultInstance;
+  }
+
+  static EchoStatus _defaultInstance;
+  static void $checkItem(EchoStatus v) {
+    if (v is! EchoStatus) checkItemFailed(v, 'EchoStatus');
+  }
+
+  int get code => $_get(0, 1, 0);
+  set code(int v) {
+    $_setUnsignedInt32(0, 1, v);
+  }
+
+  bool hasCode() => $_has(0, 1);
+  void clearCode() => clearField(1);
+
+  String get message => $_get(1, 2, '');
+  set message(String v) {
+    $_setString(1, 2, v);
+  }
+
+  bool hasMessage() => $_has(1, 2);
+  void clearMessage() => clearField(2);
+}
+
+class _ReadonlyEchoStatus extends EchoStatus with ReadonlyMessageMixin {}
+
+class SimpleRequest extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('SimpleRequest')
+    ..e/*<PayloadType>*/(1, 'responseType', PbFieldType.OE,
+        PayloadType.COMPRESSABLE, PayloadType.valueOf)
+    ..a/*<int>*/(2, 'responseSize', PbFieldType.O3)
+    ..a/*<Payload>*/(
+        3, 'payload', PbFieldType.OM, Payload.getDefault, Payload.create)
+    ..a/*<bool>*/(4, 'fillUsername', PbFieldType.OB)
+    ..a/*<bool>*/(5, 'fillOauthScope', PbFieldType.OB)
+    ..a/*<BoolValue>*/(6, 'responseCompressed', PbFieldType.OM,
+        BoolValue.getDefault, BoolValue.create)
+    ..a/*<EchoStatus>*/(7, 'responseStatus', PbFieldType.OM,
+        EchoStatus.getDefault, EchoStatus.create)
+    ..a/*<BoolValue>*/(8, 'expectCompressed', PbFieldType.OM,
+        BoolValue.getDefault, BoolValue.create)
+    ..hasRequiredFields = false;
+
+  SimpleRequest() : super();
+  SimpleRequest.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  SimpleRequest.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  SimpleRequest clone() => new SimpleRequest()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static SimpleRequest create() => new SimpleRequest();
+  static PbList<SimpleRequest> createRepeated() => new PbList<SimpleRequest>();
+  static SimpleRequest getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlySimpleRequest();
+    return _defaultInstance;
+  }
+
+  static SimpleRequest _defaultInstance;
+  static void $checkItem(SimpleRequest v) {
+    if (v is! SimpleRequest) checkItemFailed(v, 'SimpleRequest');
+  }
+
+  PayloadType get responseType => $_get(0, 1, null);
+  set responseType(PayloadType v) {
+    setField(1, v);
+  }
+
+  bool hasResponseType() => $_has(0, 1);
+  void clearResponseType() => clearField(1);
+
+  int get responseSize => $_get(1, 2, 0);
+  set responseSize(int v) {
+    $_setUnsignedInt32(1, 2, v);
+  }
+
+  bool hasResponseSize() => $_has(1, 2);
+  void clearResponseSize() => clearField(2);
+
+  Payload get payload => $_get(2, 3, null);
+  set payload(Payload v) {
+    setField(3, v);
+  }
+
+  bool hasPayload() => $_has(2, 3);
+  void clearPayload() => clearField(3);
+
+  bool get fillUsername => $_get(3, 4, false);
+  set fillUsername(bool v) {
+    $_setBool(3, 4, v);
+  }
+
+  bool hasFillUsername() => $_has(3, 4);
+  void clearFillUsername() => clearField(4);
+
+  bool get fillOauthScope => $_get(4, 5, false);
+  set fillOauthScope(bool v) {
+    $_setBool(4, 5, v);
+  }
+
+  bool hasFillOauthScope() => $_has(4, 5);
+  void clearFillOauthScope() => clearField(5);
+
+  BoolValue get responseCompressed => $_get(5, 6, null);
+  set responseCompressed(BoolValue v) {
+    setField(6, v);
+  }
+
+  bool hasResponseCompressed() => $_has(5, 6);
+  void clearResponseCompressed() => clearField(6);
+
+  EchoStatus get responseStatus => $_get(6, 7, null);
+  set responseStatus(EchoStatus v) {
+    setField(7, v);
+  }
+
+  bool hasResponseStatus() => $_has(6, 7);
+  void clearResponseStatus() => clearField(7);
+
+  BoolValue get expectCompressed => $_get(7, 8, null);
+  set expectCompressed(BoolValue v) {
+    setField(8, v);
+  }
+
+  bool hasExpectCompressed() => $_has(7, 8);
+  void clearExpectCompressed() => clearField(8);
+}
+
+class _ReadonlySimpleRequest extends SimpleRequest with ReadonlyMessageMixin {}
+
+class SimpleResponse extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('SimpleResponse')
+    ..a/*<Payload>*/(
+        1, 'payload', PbFieldType.OM, Payload.getDefault, Payload.create)
+    ..a/*<String>*/(2, 'username', PbFieldType.OS)
+    ..a/*<String>*/(3, 'oauthScope', PbFieldType.OS)
+    ..hasRequiredFields = false;
+
+  SimpleResponse() : super();
+  SimpleResponse.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  SimpleResponse.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  SimpleResponse clone() => new SimpleResponse()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static SimpleResponse create() => new SimpleResponse();
+  static PbList<SimpleResponse> createRepeated() =>
+      new PbList<SimpleResponse>();
+  static SimpleResponse getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlySimpleResponse();
+    return _defaultInstance;
+  }
+
+  static SimpleResponse _defaultInstance;
+  static void $checkItem(SimpleResponse v) {
+    if (v is! SimpleResponse) checkItemFailed(v, 'SimpleResponse');
+  }
+
+  Payload get payload => $_get(0, 1, null);
+  set payload(Payload v) {
+    setField(1, v);
+  }
+
+  bool hasPayload() => $_has(0, 1);
+  void clearPayload() => clearField(1);
+
+  String get username => $_get(1, 2, '');
+  set username(String v) {
+    $_setString(1, 2, v);
+  }
+
+  bool hasUsername() => $_has(1, 2);
+  void clearUsername() => clearField(2);
+
+  String get oauthScope => $_get(2, 3, '');
+  set oauthScope(String v) {
+    $_setString(2, 3, v);
+  }
+
+  bool hasOauthScope() => $_has(2, 3);
+  void clearOauthScope() => clearField(3);
+}
+
+class _ReadonlySimpleResponse extends SimpleResponse with ReadonlyMessageMixin {
+}
+
+class StreamingInputCallRequest extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('StreamingInputCallRequest')
+    ..a/*<Payload>*/(
+        1, 'payload', PbFieldType.OM, Payload.getDefault, Payload.create)
+    ..a/*<BoolValue>*/(2, 'expectCompressed', PbFieldType.OM,
+        BoolValue.getDefault, BoolValue.create)
+    ..hasRequiredFields = false;
+
+  StreamingInputCallRequest() : super();
+  StreamingInputCallRequest.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  StreamingInputCallRequest.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  StreamingInputCallRequest clone() =>
+      new StreamingInputCallRequest()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static StreamingInputCallRequest create() => new StreamingInputCallRequest();
+  static PbList<StreamingInputCallRequest> createRepeated() =>
+      new PbList<StreamingInputCallRequest>();
+  static StreamingInputCallRequest getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlyStreamingInputCallRequest();
+    return _defaultInstance;
+  }
+
+  static StreamingInputCallRequest _defaultInstance;
+  static void $checkItem(StreamingInputCallRequest v) {
+    if (v is! StreamingInputCallRequest)
+      checkItemFailed(v, 'StreamingInputCallRequest');
+  }
+
+  Payload get payload => $_get(0, 1, null);
+  set payload(Payload v) {
+    setField(1, v);
+  }
+
+  bool hasPayload() => $_has(0, 1);
+  void clearPayload() => clearField(1);
+
+  BoolValue get expectCompressed => $_get(1, 2, null);
+  set expectCompressed(BoolValue v) {
+    setField(2, v);
+  }
+
+  bool hasExpectCompressed() => $_has(1, 2);
+  void clearExpectCompressed() => clearField(2);
+}
+
+class _ReadonlyStreamingInputCallRequest extends StreamingInputCallRequest
+    with ReadonlyMessageMixin {}
+
+class StreamingInputCallResponse extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('StreamingInputCallResponse')
+    ..a/*<int>*/(1, 'aggregatedPayloadSize', PbFieldType.O3)
+    ..hasRequiredFields = false;
+
+  StreamingInputCallResponse() : super();
+  StreamingInputCallResponse.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  StreamingInputCallResponse.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  StreamingInputCallResponse clone() =>
+      new StreamingInputCallResponse()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static StreamingInputCallResponse create() =>
+      new StreamingInputCallResponse();
+  static PbList<StreamingInputCallResponse> createRepeated() =>
+      new PbList<StreamingInputCallResponse>();
+  static StreamingInputCallResponse getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlyStreamingInputCallResponse();
+    return _defaultInstance;
+  }
+
+  static StreamingInputCallResponse _defaultInstance;
+  static void $checkItem(StreamingInputCallResponse v) {
+    if (v is! StreamingInputCallResponse)
+      checkItemFailed(v, 'StreamingInputCallResponse');
+  }
+
+  int get aggregatedPayloadSize => $_get(0, 1, 0);
+  set aggregatedPayloadSize(int v) {
+    $_setUnsignedInt32(0, 1, v);
+  }
+
+  bool hasAggregatedPayloadSize() => $_has(0, 1);
+  void clearAggregatedPayloadSize() => clearField(1);
+}
+
+class _ReadonlyStreamingInputCallResponse extends StreamingInputCallResponse
+    with ReadonlyMessageMixin {}
+
+class ResponseParameters extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('ResponseParameters')
+    ..a/*<int>*/(1, 'size', PbFieldType.O3)
+    ..a/*<int>*/(2, 'intervalUs', PbFieldType.O3)
+    ..a/*<BoolValue>*/(
+        3, 'compressed', PbFieldType.OM, BoolValue.getDefault, BoolValue.create)
+    ..hasRequiredFields = false;
+
+  ResponseParameters() : super();
+  ResponseParameters.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  ResponseParameters.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  ResponseParameters clone() =>
+      new ResponseParameters()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static ResponseParameters create() => new ResponseParameters();
+  static PbList<ResponseParameters> createRepeated() =>
+      new PbList<ResponseParameters>();
+  static ResponseParameters getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlyResponseParameters();
+    return _defaultInstance;
+  }
+
+  static ResponseParameters _defaultInstance;
+  static void $checkItem(ResponseParameters v) {
+    if (v is! ResponseParameters) checkItemFailed(v, 'ResponseParameters');
+  }
+
+  int get size => $_get(0, 1, 0);
+  set size(int v) {
+    $_setUnsignedInt32(0, 1, v);
+  }
+
+  bool hasSize() => $_has(0, 1);
+  void clearSize() => clearField(1);
+
+  int get intervalUs => $_get(1, 2, 0);
+  set intervalUs(int v) {
+    $_setUnsignedInt32(1, 2, v);
+  }
+
+  bool hasIntervalUs() => $_has(1, 2);
+  void clearIntervalUs() => clearField(2);
+
+  BoolValue get compressed => $_get(2, 3, null);
+  set compressed(BoolValue v) {
+    setField(3, v);
+  }
+
+  bool hasCompressed() => $_has(2, 3);
+  void clearCompressed() => clearField(3);
+}
+
+class _ReadonlyResponseParameters extends ResponseParameters
+    with ReadonlyMessageMixin {}
+
+class StreamingOutputCallRequest extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('StreamingOutputCallRequest')
+    ..e/*<PayloadType>*/(1, 'responseType', PbFieldType.OE,
+        PayloadType.COMPRESSABLE, PayloadType.valueOf)
+    ..pp/*<ResponseParameters>*/(2, 'responseParameters', PbFieldType.PM,
+        ResponseParameters.$checkItem, ResponseParameters.create)
+    ..a/*<Payload>*/(
+        3, 'payload', PbFieldType.OM, Payload.getDefault, Payload.create)
+    ..a/*<EchoStatus>*/(7, 'responseStatus', PbFieldType.OM,
+        EchoStatus.getDefault, EchoStatus.create)
+    ..hasRequiredFields = false;
+
+  StreamingOutputCallRequest() : super();
+  StreamingOutputCallRequest.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  StreamingOutputCallRequest.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  StreamingOutputCallRequest clone() =>
+      new StreamingOutputCallRequest()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static StreamingOutputCallRequest create() =>
+      new StreamingOutputCallRequest();
+  static PbList<StreamingOutputCallRequest> createRepeated() =>
+      new PbList<StreamingOutputCallRequest>();
+  static StreamingOutputCallRequest getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlyStreamingOutputCallRequest();
+    return _defaultInstance;
+  }
+
+  static StreamingOutputCallRequest _defaultInstance;
+  static void $checkItem(StreamingOutputCallRequest v) {
+    if (v is! StreamingOutputCallRequest)
+      checkItemFailed(v, 'StreamingOutputCallRequest');
+  }
+
+  PayloadType get responseType => $_get(0, 1, null);
+  set responseType(PayloadType v) {
+    setField(1, v);
+  }
+
+  bool hasResponseType() => $_has(0, 1);
+  void clearResponseType() => clearField(1);
+
+  List<ResponseParameters> get responseParameters => $_get(1, 2, null);
+
+  Payload get payload => $_get(2, 3, null);
+  set payload(Payload v) {
+    setField(3, v);
+  }
+
+  bool hasPayload() => $_has(2, 3);
+  void clearPayload() => clearField(3);
+
+  EchoStatus get responseStatus => $_get(3, 7, null);
+  set responseStatus(EchoStatus v) {
+    setField(7, v);
+  }
+
+  bool hasResponseStatus() => $_has(3, 7);
+  void clearResponseStatus() => clearField(7);
+}
+
+class _ReadonlyStreamingOutputCallRequest extends StreamingOutputCallRequest
+    with ReadonlyMessageMixin {}
+
+class StreamingOutputCallResponse extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('StreamingOutputCallResponse')
+    ..a/*<Payload>*/(
+        1, 'payload', PbFieldType.OM, Payload.getDefault, Payload.create)
+    ..hasRequiredFields = false;
+
+  StreamingOutputCallResponse() : super();
+  StreamingOutputCallResponse.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  StreamingOutputCallResponse.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  StreamingOutputCallResponse clone() =>
+      new StreamingOutputCallResponse()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static StreamingOutputCallResponse create() =>
+      new StreamingOutputCallResponse();
+  static PbList<StreamingOutputCallResponse> createRepeated() =>
+      new PbList<StreamingOutputCallResponse>();
+  static StreamingOutputCallResponse getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlyStreamingOutputCallResponse();
+    return _defaultInstance;
+  }
+
+  static StreamingOutputCallResponse _defaultInstance;
+  static void $checkItem(StreamingOutputCallResponse v) {
+    if (v is! StreamingOutputCallResponse)
+      checkItemFailed(v, 'StreamingOutputCallResponse');
+  }
+
+  Payload get payload => $_get(0, 1, null);
+  set payload(Payload v) {
+    setField(1, v);
+  }
+
+  bool hasPayload() => $_has(0, 1);
+  void clearPayload() => clearField(1);
+}
+
+class _ReadonlyStreamingOutputCallResponse extends StreamingOutputCallResponse
+    with ReadonlyMessageMixin {}
+
+class ReconnectParams extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('ReconnectParams')
+    ..a/*<int>*/(1, 'maxReconnectBackoffMs', PbFieldType.O3)
+    ..hasRequiredFields = false;
+
+  ReconnectParams() : super();
+  ReconnectParams.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  ReconnectParams.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  ReconnectParams clone() => new ReconnectParams()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static ReconnectParams create() => new ReconnectParams();
+  static PbList<ReconnectParams> createRepeated() =>
+      new PbList<ReconnectParams>();
+  static ReconnectParams getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlyReconnectParams();
+    return _defaultInstance;
+  }
+
+  static ReconnectParams _defaultInstance;
+  static void $checkItem(ReconnectParams v) {
+    if (v is! ReconnectParams) checkItemFailed(v, 'ReconnectParams');
+  }
+
+  int get maxReconnectBackoffMs => $_get(0, 1, 0);
+  set maxReconnectBackoffMs(int v) {
+    $_setUnsignedInt32(0, 1, v);
+  }
+
+  bool hasMaxReconnectBackoffMs() => $_has(0, 1);
+  void clearMaxReconnectBackoffMs() => clearField(1);
+}
+
+class _ReadonlyReconnectParams extends ReconnectParams
+    with ReadonlyMessageMixin {}
+
+class ReconnectInfo extends GeneratedMessage {
+  static final BuilderInfo _i = new BuilderInfo('ReconnectInfo')
+    ..a/*<bool>*/(1, 'passed', PbFieldType.OB)
+    ..p/*<int>*/(2, 'backoffMs', PbFieldType.P3)
+    ..hasRequiredFields = false;
+
+  ReconnectInfo() : super();
+  ReconnectInfo.fromBuffer(List<int> i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromBuffer(i, r);
+  ReconnectInfo.fromJson(String i,
+      [ExtensionRegistry r = ExtensionRegistry.EMPTY])
+      : super.fromJson(i, r);
+  ReconnectInfo clone() => new ReconnectInfo()..mergeFromMessage(this);
+  BuilderInfo get info_ => _i;
+  static ReconnectInfo create() => new ReconnectInfo();
+  static PbList<ReconnectInfo> createRepeated() => new PbList<ReconnectInfo>();
+  static ReconnectInfo getDefault() {
+    if (_defaultInstance == null)
+      _defaultInstance = new _ReadonlyReconnectInfo();
+    return _defaultInstance;
+  }
+
+  static ReconnectInfo _defaultInstance;
+  static void $checkItem(ReconnectInfo v) {
+    if (v is! ReconnectInfo) checkItemFailed(v, 'ReconnectInfo');
+  }
+
+  bool get passed => $_get(0, 1, false);
+  set passed(bool v) {
+    $_setBool(0, 1, v);
+  }
+
+  bool hasPassed() => $_has(0, 1);
+  void clearPassed() => clearField(1);
+
+  List<int> get backoffMs => $_get(1, 2, null);
+}
+
+class _ReadonlyReconnectInfo extends ReconnectInfo with ReadonlyMessageMixin {}

interop/lib/src/generated/messages.pbenum.dart

@@ -0,0 +1,27 @@
+///
+//  Generated code. Do not modify.
+///
+// ignore_for_file: non_constant_identifier_names
+// ignore_for_file: library_prefixes
+library grpc.testing_messages_pbenum;
+
+// ignore: UNUSED_SHOWN_NAME
+import 'dart:core' show int, dynamic, String, List, Map;
+import 'package:protobuf/protobuf.dart';
+
+class PayloadType extends ProtobufEnum {
+  static const PayloadType COMPRESSABLE =
+      const PayloadType._(0, 'COMPRESSABLE');
+
+  static const List<PayloadType> values = const <PayloadType>[
+    COMPRESSABLE,
+  ];
+
+  static final Map<int, dynamic> _byValue = ProtobufEnum.initByValue(values);
+  static PayloadType valueOf(int value) => _byValue[value] as PayloadType;
+  static void $checkItem(PayloadType v) {
+    if (v is! PayloadType) checkItemFailed(v, 'PayloadType');
+  }
+
+  const PayloadType._(int v, String n) : super(v, n);
+}

interop/lib/src/generated/test.pb.dart

@@ -0,0 +1,9 @@
+///
+//  Generated code. Do not modify.
+///
+// ignore_for_file: non_constant_identifier_names
+// ignore_for_file: library_prefixes
+library grpc.testing_test;
+
+// ignore: UNUSED_SHOWN_NAME
+import 'dart:core' show int, bool, double, String, List, override;

interop/lib/src/generated/test.pbgrpc.dart

@@ -0,0 +1,330 @@
+///
+//  Generated code. Do not modify.
+///
+// ignore_for_file: non_constant_identifier_names
+// ignore_for_file: library_prefixes
+library grpc.testing_test_pbgrpc;
+
+import 'dart:async';
+
+import 'package:grpc/grpc.dart';
+
+import 'empty.pb.dart';
+import 'messages.pb.dart';
+export 'test.pb.dart';
+
+class TestServiceClient {
+  final ClientChannel _channel;
+
+  static final _$emptyCall = new ClientMethod<Empty, Empty>(
+      '/grpc.testing.TestService/EmptyCall',
+      (Empty value) => value.writeToBuffer(),
+      (List<int> value) => new Empty.fromBuffer(value));
+  static final _$unaryCall = new ClientMethod<SimpleRequest, SimpleResponse>(
+      '/grpc.testing.TestService/UnaryCall',
+      (SimpleRequest value) => value.writeToBuffer(),
+      (List<int> value) => new SimpleResponse.fromBuffer(value));
+  static final _$cacheableUnaryCall =
+      new ClientMethod<SimpleRequest, SimpleResponse>(
+          '/grpc.testing.TestService/CacheableUnaryCall',
+          (SimpleRequest value) => value.writeToBuffer(),
+          (List<int> value) => new SimpleResponse.fromBuffer(value));
+  static final _$streamingOutputCall =
+      new ClientMethod<StreamingOutputCallRequest, StreamingOutputCallResponse>(
+          '/grpc.testing.TestService/StreamingOutputCall',
+          (StreamingOutputCallRequest value) => value.writeToBuffer(),
+          (List<int> value) =>
+              new StreamingOutputCallResponse.fromBuffer(value));
+  static final _$streamingInputCall =
+      new ClientMethod<StreamingInputCallRequest, StreamingInputCallResponse>(
+          '/grpc.testing.TestService/StreamingInputCall',
+          (StreamingInputCallRequest value) => value.writeToBuffer(),
+          (List<int> value) =>
+              new StreamingInputCallResponse.fromBuffer(value));
+  static final _$fullDuplexCall =
+      new ClientMethod<StreamingOutputCallRequest, StreamingOutputCallResponse>(
+          '/grpc.testing.TestService/FullDuplexCall',
+          (StreamingOutputCallRequest value) => value.writeToBuffer(),
+          (List<int> value) =>
+              new StreamingOutputCallResponse.fromBuffer(value));
+  static final _$halfDuplexCall =
+      new ClientMethod<StreamingOutputCallRequest, StreamingOutputCallResponse>(
+          '/grpc.testing.TestService/HalfDuplexCall',
+          (StreamingOutputCallRequest value) => value.writeToBuffer(),
+          (List<int> value) =>
+              new StreamingOutputCallResponse.fromBuffer(value));
+  static final _$unimplementedCall = new ClientMethod<Empty, Empty>(
+      '/grpc.testing.TestService/UnimplementedCall',
+      (Empty value) => value.writeToBuffer(),
+      (List<int> value) => new Empty.fromBuffer(value));
+
+  TestServiceClient(this._channel);
+
+  ResponseFuture<Empty> emptyCall(Empty request, {CallOptions options}) {
+    final call = new ClientCall(_channel, _$emptyCall, options: options);
+    call.request
+      ..add(request)
+      ..close();
+    return new ResponseFuture(call);
+  }
+
+  ResponseFuture<SimpleResponse> unaryCall(SimpleRequest request,
+      {CallOptions options}) {
+    final call = new ClientCall(_channel, _$unaryCall, options: options);
+    call.request
+      ..add(request)
+      ..close();
+    return new ResponseFuture(call);
+  }
+
+  ResponseFuture<SimpleResponse> cacheableUnaryCall(SimpleRequest request,
+      {CallOptions options}) {
+    final call =
+        new ClientCall(_channel, _$cacheableUnaryCall, options: options);
+    call.request
+      ..add(request)
+      ..close();
+    return new ResponseFuture(call);
+  }
+
+  ResponseStream<StreamingOutputCallResponse> streamingOutputCall(
+      StreamingOutputCallRequest request,
+      {CallOptions options}) {
+    final call =
+        new ClientCall(_channel, _$streamingOutputCall, options: options);
+    call.request
+      ..add(request)
+      ..close();
+    return new ResponseStream(call);
+  }
+
+  ResponseFuture<StreamingInputCallResponse> streamingInputCall(
+      Stream<StreamingInputCallRequest> request,
+      {CallOptions options}) {
+    final call =
+        new ClientCall(_channel, _$streamingInputCall, options: options);
+    request.pipe(call.request);
+    return new ResponseFuture(call);
+  }
+
+  ResponseStream<StreamingOutputCallResponse> fullDuplexCall(
+      Stream<StreamingOutputCallRequest> request,
+      {CallOptions options}) {
+    final call = new ClientCall(_channel, _$fullDuplexCall, options: options);
+    request.pipe(call.request);
+    return new ResponseStream(call);
+  }
+
+  ResponseStream<StreamingOutputCallResponse> halfDuplexCall(
+      Stream<StreamingOutputCallRequest> request,
+      {CallOptions options}) {
+    final call = new ClientCall(_channel, _$halfDuplexCall, options: options);
+    request.pipe(call.request);
+    return new ResponseStream(call);
+  }
+
+  ResponseFuture<Empty> unimplementedCall(Empty request,
+      {CallOptions options}) {
+    final call =
+        new ClientCall(_channel, _$unimplementedCall, options: options);
+    call.request
+      ..add(request)
+      ..close();
+    return new ResponseFuture(call);
+  }
+}
+
+abstract class TestServiceBase extends Service {
+  String get $name => 'grpc.testing.TestService';
+
+  TestServiceBase() {
+    $addMethod(new ServiceMethod(
+        'EmptyCall',
+        emptyCall_Pre,
+        false,
+        false,
+        (List<int> value) => new Empty.fromBuffer(value),
+        (Empty value) => value.writeToBuffer()));
+    $addMethod(new ServiceMethod(
+        'UnaryCall',
+        unaryCall_Pre,
+        false,
+        false,
+        (List<int> value) => new SimpleRequest.fromBuffer(value),
+        (SimpleResponse value) => value.writeToBuffer()));
+    $addMethod(new ServiceMethod(
+        'CacheableUnaryCall',
+        cacheableUnaryCall_Pre,
+        false,
+        false,
+        (List<int> value) => new SimpleRequest.fromBuffer(value),
+        (SimpleResponse value) => value.writeToBuffer()));
+    $addMethod(new ServiceMethod(
+        'StreamingOutputCall',
+        streamingOutputCall_Pre,
+        false,
+        true,
+        (List<int> value) => new StreamingOutputCallRequest.fromBuffer(value),
+        (StreamingOutputCallResponse value) => value.writeToBuffer()));
+    $addMethod(new ServiceMethod(
+        'StreamingInputCall',
+        streamingInputCall,
+        true,
+        false,
+        (List<int> value) => new StreamingInputCallRequest.fromBuffer(value),
+        (StreamingInputCallResponse value) => value.writeToBuffer()));
+    $addMethod(new ServiceMethod(
+        'FullDuplexCall',
+        fullDuplexCall,
+        true,
+        true,
+        (List<int> value) => new StreamingOutputCallRequest.fromBuffer(value),
+        (StreamingOutputCallResponse value) => value.writeToBuffer()));
+    $addMethod(new ServiceMethod(
+        'HalfDuplexCall',
+        halfDuplexCall,
+        true,
+        true,
+        (List<int> value) => new StreamingOutputCallRequest.fromBuffer(value),
+        (StreamingOutputCallResponse value) => value.writeToBuffer()));
+  }
+
+  Future<Empty> emptyCall_Pre(ServiceCall call, Future<Empty> request) async {
+    return emptyCall(call, await request);
+  }
+
+  Future<SimpleResponse> unaryCall_Pre(
+      ServiceCall call, Future<SimpleRequest> request) async {
+    return unaryCall(call, await request);
+  }
+
+  Future<SimpleResponse> cacheableUnaryCall_Pre(
+      ServiceCall call, Future<SimpleRequest> request) async {
+    return cacheableUnaryCall(call, await request);
+  }
+
+  Stream<StreamingOutputCallResponse> streamingOutputCall_Pre(
+      ServiceCall call, Future<StreamingOutputCallRequest> request) async* {
+    yield* streamingOutputCall(call, await request);
+  }
+
+  Future<Empty> emptyCall(ServiceCall call, Empty request);
+  Future<SimpleResponse> unaryCall(ServiceCall call, SimpleRequest request);
+  Future<SimpleResponse> cacheableUnaryCall(
+      ServiceCall call, SimpleRequest request);
+  Stream<StreamingOutputCallResponse> streamingOutputCall(
+      ServiceCall call, StreamingOutputCallRequest request);
+  Future<StreamingInputCallResponse> streamingInputCall(
+      ServiceCall call, Stream<StreamingInputCallRequest> request);
+  Stream<StreamingOutputCallResponse> fullDuplexCall(
+      ServiceCall call, Stream<StreamingOutputCallRequest> request);
+  Stream<StreamingOutputCallResponse> halfDuplexCall(
+      ServiceCall call, Stream<StreamingOutputCallRequest> request);
+}
+
+class UnimplementedServiceClient {
+  final ClientChannel _channel;
+
+  static final _$unimplementedCall = new ClientMethod<Empty, Empty>(
+      '/grpc.testing.UnimplementedService/UnimplementedCall',
+      (Empty value) => value.writeToBuffer(),
+      (List<int> value) => new Empty.fromBuffer(value));
+
+  UnimplementedServiceClient(this._channel);
+
+  ResponseFuture<Empty> unimplementedCall(Empty request,
+      {CallOptions options}) {
+    final call =
+        new ClientCall(_channel, _$unimplementedCall, options: options);
+    call.request
+      ..add(request)
+      ..close();
+    return new ResponseFuture(call);
+  }
+}
+
+abstract class UnimplementedServiceBase extends Service {
+  String get $name => 'grpc.testing.UnimplementedService';
+
+  UnimplementedServiceBase() {
+    $addMethod(new ServiceMethod(
+        'UnimplementedCall',
+        unimplementedCall_Pre,
+        false,
+        false,
+        (List<int> value) => new Empty.fromBuffer(value),
+        (Empty value) => value.writeToBuffer()));
+  }
+
+  Future<Empty> unimplementedCall_Pre(
+      ServiceCall call, Future<Empty> request) async {
+    return unimplementedCall(call, await request);
+  }
+
+  Future<Empty> unimplementedCall(ServiceCall call, Empty request);
+}
+
+class ReconnectServiceClient {
+  final ClientChannel _channel;
+
+  static final _$start = new ClientMethod<ReconnectParams, Empty>(
+      '/grpc.testing.ReconnectService/Start',
+      (ReconnectParams value) => value.writeToBuffer(),
+      (List<int> value) => new Empty.fromBuffer(value));
+  static final _$stop = new ClientMethod<Empty, ReconnectInfo>(
+      '/grpc.testing.ReconnectService/Stop',
+      (Empty value) => value.writeToBuffer(),
+      (List<int> value) => new ReconnectInfo.fromBuffer(value));
+
+  ReconnectServiceClient(this._channel);
+
+  ResponseFuture<Empty> start(ReconnectParams request, {CallOptions options}) {
+    final call = new ClientCall(_channel, _$start, options: options);
+    call.request
+      ..add(request)
+      ..close();
+    return new ResponseFuture(call);
+  }
+
+  ResponseFuture<ReconnectInfo> stop(Empty request, {CallOptions options}) {
+    final call = new ClientCall(_channel, _$stop, options: options);
+    call.request
+      ..add(request)
+      ..close();
+    return new ResponseFuture(call);
+  }
+}
+
+abstract class ReconnectServiceBase extends Service {
+  String get $name => 'grpc.testing.ReconnectService';
+
+  ReconnectServiceBase() {
+    $addMethod(new ServiceMethod(
+        'Start',
+        start_Pre,
+        false,
+        false,
+        (List<int> value) => new ReconnectParams.fromBuffer(value),
+        (Empty value) => value.writeToBuffer()));
+    $addMethod(new ServiceMethod(
+        'Stop',
+        stop_Pre,
+        false,
+        false,
+        (List<int> value) => new Empty.fromBuffer(value),
+        (ReconnectInfo value) => value.writeToBuffer()));
+  }
+
+  Future<Empty> start_Pre(
+      ServiceCall call, Future<ReconnectParams> request) async {
+    return start(call, await request);
+  }
+
+  Future<ReconnectInfo> stop_Pre(
+      ServiceCall call, Future<Empty> request) async {
+    return stop(call, await request);
+  }
+
+  Future<Empty> start(ServiceCall call, ReconnectParams request);
+  Future<ReconnectInfo> stop(ServiceCall call, Empty request);
+}

interop/protos/empty.proto

@@ -0,0 +1,43 @@
+
+// Copyright 2015, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//     * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+syntax = "proto3";
+
+package grpc.testing;
+
+// An empty message that you can re-use to avoid defining duplicated empty
+// messages in your project. A typical example is to use it as argument or the
+// return value of a service API. For instance:
+//
+//   service Foo {
+//     rpc Bar (grpc.testing.Empty) returns (grpc.testing.Empty) { };
+//   };
+//
+message Empty {}

interop/protos/messages.proto

@@ -0,0 +1,184 @@
+
+// Copyright 2015-2016, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//     * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+// Message definitions to be used by integration test service definitions.
+
+syntax = "proto3";
+
+package grpc.testing;
+
+// TODO(dgq): Go back to using well-known types once
+// https://github.com/grpc/grpc/issues/6980 has been fixed.
+// import "google/protobuf/wrappers.proto";
+message BoolValue {
+  // The bool value.
+  bool value = 1;
+}
+
+// DEPRECATED, don't use. To be removed shortly.
+// The type of payload that should be returned.
+enum PayloadType {
+  // Compressable text format.
+  COMPRESSABLE = 0;
+}
+
+// A block of data, to simply increase gRPC message size.
+message Payload {
+  // DEPRECATED, don't use. To be removed shortly.
+  // The type of data in body.
+  PayloadType type = 1;
+  // Primary contents of payload.
+  bytes body = 2;
+}
+
+// A protobuf representation for grpc status. This is used by test
+// clients to specify a status that the server should attempt to return.
+message EchoStatus {
+  int32 code = 1;
+  string message = 2;
+}
+
+// Unary request.
+message SimpleRequest {
+  // DEPRECATED, don't use. To be removed shortly.
+  // Desired payload type in the response from the server.
+  // If response_type is RANDOM, server randomly chooses one from other formats.
+  PayloadType response_type = 1;
+
+  // Desired payload size in the response from the server.
+  int32 response_size = 2;
+
+  // Optional input payload sent along with the request.
+  Payload payload = 3;
+
+  // Whether SimpleResponse should include username.
+  bool fill_username = 4;
+
+  // Whether SimpleResponse should include OAuth scope.
+  bool fill_oauth_scope = 5;
+
+  // Whether to request the server to compress the response. This field is
+  // "nullable" in order to interoperate seamlessly with clients not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the response's compression status.
+  BoolValue response_compressed = 6;
+
+  // Whether server should return a given status
+  EchoStatus response_status = 7;
+
+  // Whether the server should expect this request to be compressed.
+  BoolValue expect_compressed = 8;
+}
+
+// Unary response, as configured by the request.
+message SimpleResponse {
+  // Payload to increase message size.
+  Payload payload = 1;
+  // The user the request came from, for verifying authentication was
+  // successful when the client expected it.
+  string username = 2;
+  // OAuth scope.
+  string oauth_scope = 3;
+}
+
+// Client-streaming request.
+message StreamingInputCallRequest {
+  // Optional input payload sent along with the request.
+  Payload payload = 1;
+
+  // Whether the server should expect this request to be compressed. This field
+  // is "nullable" in order to interoperate seamlessly with servers not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the request's compression status.
+  BoolValue expect_compressed = 2;
+
+  // Not expecting any payload from the response.
+}
+
+// Client-streaming response.
+message StreamingInputCallResponse {
+  // Aggregated size of payloads received from the client.
+  int32 aggregated_payload_size = 1;
+}
+
+// Configuration for a particular response.
+message ResponseParameters {
+  // Desired payload sizes in responses from the server.
+  int32 size = 1;
+
+  // Desired interval between consecutive responses in the response stream in
+  // microseconds.
+  int32 interval_us = 2;
+
+  // Whether to request the server to compress the response. This field is
+  // "nullable" in order to interoperate seamlessly with clients not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the response's compression status.
+  BoolValue compressed = 3;
+}
+
+// Server-streaming request.
+message StreamingOutputCallRequest {
+  // DEPRECATED, don't use. To be removed shortly.
+  // Desired payload type in the response from the server.
+  // If response_type is RANDOM, the payload from each response in the stream
+  // might be of different types. This is to simulate a mixed type of payload
+  // stream.
+  PayloadType response_type = 1;
+
+  // Configuration for each expected response message.
+  repeated ResponseParameters response_parameters = 2;
+
+  // Optional input payload sent along with the request.
+  Payload payload = 3;
+
+  // Whether server should return a given status.
+  EchoStatus response_status = 7;
+}
+
+// Server-streaming response, as configured by the request and parameters.
+message StreamingOutputCallResponse {
+  // Payload to increase response size.
+  Payload payload = 1;
+}
+
+// For reconnect interop test only.
+// Client tells server what reconnection parameters it used.
+message ReconnectParams {
+  int32 max_reconnect_backoff_ms = 1;
+}
+
+// For reconnect interop test only.
+// Server tells client whether its reconnects are following the spec and the
+// reconnect backoffs it saw.
+message ReconnectInfo {
+  bool passed = 1;
+  repeated int32 backoff_ms = 2;
+}

interop/protos/test.proto

@@ -0,0 +1,94 @@
+
+// Copyright 2015-2016, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//     * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+// An integration test service that covers all the method signature permutations
+// of unary/streaming requests/responses.
+
+syntax = "proto3";
+
+import "empty.proto";
+import "messages.proto";
+
+package grpc.testing;
+
+// A simple service to test the various types of RPCs and experiment with
+// performance with various types of payload.
+service TestService {
+  // One empty request followed by one empty response.
+  rpc EmptyCall(grpc.testing.Empty) returns (grpc.testing.Empty);
+
+  // One request followed by one response.
+  rpc UnaryCall(SimpleRequest) returns (SimpleResponse);
+
+  // One request followed by one response. Response has cache control
+  // headers set such that a caching HTTP proxy (such as GFE) can
+  // satisfy subsequent requests.
+  rpc CacheableUnaryCall(SimpleRequest) returns (SimpleResponse);
+
+  // One request followed by a sequence of responses (streamed download).
+  // The server returns the payload with client desired type and sizes.
+  rpc StreamingOutputCall(StreamingOutputCallRequest)
+      returns (stream StreamingOutputCallResponse);
+
+  // A sequence of requests followed by one response (streamed upload).
+  // The server returns the aggregated size of client payload as the result.
+  rpc StreamingInputCall(stream StreamingInputCallRequest)
+      returns (StreamingInputCallResponse);
+
+  // A sequence of requests with each request served by the server immediately.
+  // As one request could lead to multiple responses, this interface
+  // demonstrates the idea of full duplexing.
+  rpc FullDuplexCall(stream StreamingOutputCallRequest)
+      returns (stream StreamingOutputCallResponse);
+
+  // A sequence of requests followed by a sequence of responses.
+  // The server buffers all the client requests and then serves them in order. A
+  // stream of responses are returned to the client when the server starts with
+  // first request.
+  rpc HalfDuplexCall(stream StreamingOutputCallRequest)
+      returns (stream StreamingOutputCallResponse);
+
+  // The test server will not implement this method. It will be used
+  // to test the behavior when clients call unimplemented methods.
+  rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty);
+}
+
+// A simple service NOT implemented at servers so clients can test for
+// that case.
+service UnimplementedService {
+  // A call that no server should implement
+  rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty);
+}
+
+// A service used to control reconnect server.
+service ReconnectService {
+  rpc Start(grpc.testing.ReconnectParams) returns (grpc.testing.Empty);
+  rpc Stop(grpc.testing.Empty) returns (grpc.testing.ReconnectInfo);
+}

interop/pubspec.yaml

@@ -0,0 +1,18 @@
+name: interop
+description: Dart gRPC interoperability test suite.
+version: 0.0.1
+homepage: https://github.com/dart-lang/grpc-dart
+
+environment:
+  sdk: '>=1.20.1 <2.0.0'
+
+dependencies:
+  args: ^0.13.0
+  async: ^1.13.3
+  grpc:
+    path: ../
+  protobuf: ^0.5.4
+  http2: ^0.1.2
+
+dev_dependencies:
+  test: ^0.12.0

interop/tool/regenerate.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+mkdir -p lib/src/generated
+protoc --dart_out=grpc:lib/src/generated -Iprotos/ protos/*.proto
+rm lib/src/generated/*.pbjson.dart
+rm lib/src/generated/{empty,test}.pbenum.dart
+dartfmt -w lib/src/generated

lib/grpc.dart

@@ -5,3 +5,5 @@
 export 'src/client.dart';
 export 'src/server.dart';
 export 'src/shared.dart';
+export 'src/status.dart';
+export 'src/streams.dart';

lib/src/server.dart

@@ -41,6 +41,12 @@ abstract class Service {
     _$methods[method.name] = method;
   }
 
+  /// Client metadata handler.
+  ///
+  /// Services can override this method to provide common handling of incoming
+  /// metadata from the client.
+  void $onMetadata(ServiceCall context) {}
+
   ServiceMethod $lookupMethod(String name) => _$methods[name];
 }
 
@@ -61,8 +67,7 @@ class Server {
     _services[service.$name] = service;
   }
 
-  ServiceMethod _lookupMethod(String service, String method) =>
+  Service lookupService(String service) => _services[service];
-      _services[service]?.$lookupMethod(method);
 
   Future<Null> serve() async {
     // TODO(dart-lang/grpc-dart#4): Add TLS support.
@@ -72,7 +77,7 @@ class Server {
       final connection = new ServerTransportConnection.viaSocket(socket);
       _connections.add(connection);
       connection.incomingStreams.listen((stream) {
-        new ServerHandler(_lookupMethod, stream).handle();
+        new ServerHandler(lookupService, stream).handle();
       }, onError: (error) {
         print('Connection error: $error');
       }, onDone: () {
@@ -133,7 +138,6 @@ class ServiceCall {
   ///
   /// The call will be closed after calling this method, and no further
   /// responses can be sent.
-  /// responses can be sent.
   void sendTrailer(int status, [String statusMessage]) =>
       _handler._sendTrailers(status: status, message: statusMessage);
 }
@@ -141,13 +145,15 @@ class ServiceCall {
 /// Handles an incoming gRPC call.
 class ServerHandler {
   final ServerTransportStream _stream;
-  final ServiceMethod Function(String service, String method) _methodLookup;
+  final Service Function(String service) _serviceLookup;
 
   StreamSubscription<GrpcMessage> _incomingSubscription;
 
-  Map<String, String> _clientMetadata;
+  Service _service;
   ServiceMethod _descriptor;
 
+  Map<String, String> _clientMetadata;
+
   StreamController _requests;
   bool _hasReceivedRequest = false;
 
@@ -161,7 +167,7 @@ class ServerHandler {
   DateTime _deadline;
   bool _isCanceled = false;
 
-  ServerHandler(this._methodLookup, this._stream);
+  ServerHandler(this._serviceLookup, this._stream);
 
   bool get isCanceled => _isCanceled;
   bool get _isTimedOut => _deadline?.isBefore(new DateTime.now()) ?? false;
@@ -169,7 +175,7 @@ class ServerHandler {
   void handle() {
     _stream.onTerminated = (int errorCode) {
       _isCanceled = true;
-      _responseSubscription?.cancel();
+      _cancelResponseSubscription();
     };
 
     _incomingSubscription = _stream.incomingMessages
@@ -179,6 +185,14 @@ class ServerHandler {
             onError: _onError, onDone: _onDoneError, cancelOnError: true);
   }
 
+  /// Cancel response subscription, if active. If the stream exits with an
+  /// error, just ignore it. The client is long gone, so it doesn't care.
+  /// We need the catchError() handler here, since otherwise the error would
+  /// be an unhandled exception.
+  void _cancelResponseSubscription() {
+    _responseSubscription?.cancel()?.catchError((_) {});
+  }
+
   // -- Idle state, incoming data --
 
   void _onDataIdle(GrpcMessage message) {
@@ -189,17 +203,20 @@ class ServerHandler {
     final headerMessage = message
         as GrpcMetadata; // TODO(jakobr): Cast should not be necessary here.
     _clientMetadata = headerMessage.metadata;
-    final path = _clientMetadata[':path'].split('/');
+    final path = _clientMetadata[':path'];
-    if (path.length < 3) {
+    final pathSegments = path.split('/');
+    if (pathSegments.length < 3) {
       _sendError(new GrpcError.unimplemented('Invalid path'));
+      _sinkIncoming();
       return;
     }
-    final service = path[1];
+    final serviceName = pathSegments[1];
-    final method = path[2];
+    final methodName = pathSegments[2];
-    _descriptor = _methodLookup(service, method);
+    _service = _serviceLookup(serviceName);
+    _descriptor = _service?.$lookupMethod(methodName);
     if (_descriptor == null) {
-      _sendError(
+      _sendError(new GrpcError.unimplemented('Path $path not found'));
-          new GrpcError.unimplemented('Path /$service/$method not found'));
+      _sinkIncoming();
       return;
     }
     _startStreamingRequest();
@@ -214,6 +231,7 @@ class ServerHandler {
     _incomingSubscription.onData(_onDataActive);
 
     final context = new ServiceCall(this);
+    _service.$onMetadata(context);
     if (_descriptor.streamingResponse) {
       if (_descriptor.streamingRequest) {
         _responses = _descriptor.handler(context, _requests.stream);
@@ -284,15 +302,17 @@ class ServerHandler {
       final bytes = _descriptor.responseSerializer(response);
       _stream.sendData(GrpcHttpEncoder.frame(bytes));
     } catch (error) {
-      _responseSubscription.cancel();
+      final grpcError =
+          new GrpcError.internal('Error sending response: $error');
       if (!_requests.isClosed) {
         // If we can, alert the handler that things are going wrong.
         _requests
-            .addError(new GrpcError.internal('Error sending response: $error'));
+          ..addError(grpcError)
-        _requests.close();
+          ..close();
       }
-      _incomingSubscription.cancel();
+      _sendError(grpcError);
-      _stream.terminate();
+      _cancelResponseSubscription();
+      _sinkIncoming();
     }
   }
 
@@ -348,8 +368,7 @@ class ServerHandler {
         .forEach((key, value) => trailers.add(new Header.ascii(key, value)));
     _stream.sendHeaders(trailers, endStream: true);
     // We're done!
-    _incomingSubscription.cancel();
+    _cancelResponseSubscription();
-    _responseSubscription?.cancel();
   }
 
   // -- All states, incoming error / stream closed --
@@ -358,12 +377,15 @@ class ServerHandler {
     // Exception from the incoming stream. Most likely a cancel request from the
     // client, so we treat it as such.
     _isCanceled = true;
-    _requests.addError(new GrpcError.cancelled('Cancelled'));
+    if (!_requests.isClosed) {
-    _responseSubscription?.cancel();
+      _requests.addError(new GrpcError.cancelled('Cancelled'));
+    }
+    _cancelResponseSubscription();
   }
 
   void _onDoneError() {
     _sendError(new GrpcError.unavailable('Request stream closed unexpectedly'));
+    _onDone();
   }
 
   void _onDoneExpected() {
@@ -372,10 +394,22 @@ class ServerHandler {
       _sendError(error);
       _requests.addError(error);
     }
-    _requests.close();
+    _onDone();
+  }
+
+  void _onDone() {
+    _requests?.close();
     _incomingSubscription.cancel();
   }
 
+  /// Sink incoming requests. This is used when an error has already been
+  /// reported, but we still need to consume the request stream from the client.
+  void _sinkIncoming() {
+    _incomingSubscription
+      ..onData((_) {})
+      ..onDone(_onDone);
+  }
+
   void _sendError(GrpcError error) {
     _sendTrailers(status: error.code, message: error.message);
   }