fix: fix headers not completing when call is terminated (grpc#728)

Fixes grpc#727

Author: c-lucera-pvotal

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 4.0.1
+
+* Fix header and trailing not completing if the call is terminated. Fixes [#727](https://github.com/grpc/grpc-dart/issues/727)
+
 ## 4.0.0
 
 * Set compressed flag correctly for grpc-encoding = identity. Fixes [#669](https://github.com/grpc/grpc-dart/issues/669) (https://github.com/grpc/grpc-dart/pull/693)

lib/src/client/call.dart

@@ -483,6 +483,12 @@ class ClientCall<Q, R> implements Response {
     if (_responseSubscription != null) {
       futures.add(_responseSubscription!.cancel());
     }
+    if (!_headers.isCompleted) {
+      _headers.complete({});
+    }
+    if (!_trailers.isCompleted) {
+      _trailers.complete({});
+    }
     await Future.wait(futures);
   }
 

pubspec.yaml

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

test/client_tests/call_test.dart

@@ -13,10 +13,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+import 'package:grpc/grpc.dart';
 import 'package:grpc/src/client/call.dart';
 import 'package:test/test.dart';
 
+import '../src/client_utils.dart';
+
 void main() {
+  const dummyValue = 0;
+  const cancelDurationMillis = 300;
+
+  late ClientHarness harness;
+
+  setUp(() {
+    harness = ClientHarness()..setUp();
+  });
+
+  tearDown(() {
+    harness.tearDown();
+  });
+
   test('WebCallOptions mergeWith CallOptions returns WebCallOptions', () {
     final options =
         WebCallOptions(bypassCorsPreflight: true, withCredentials: true);
@@ -28,4 +44,57 @@ void main() {
     expect(mergedOptions.bypassCorsPreflight, true);
     expect(mergedOptions.withCredentials, true);
   });
+
+  test(
+    'Cancelling a call correctly complete headers future',
+    () async {
+      final clientCall = harness.client.unary(dummyValue);
+
+      Future.delayed(
+        Duration(milliseconds: cancelDurationMillis),
+      ).then((_) => clientCall.cancel());
+
+      expect(await clientCall.headers, isEmpty);
+
+      await expectLater(
+        clientCall,
+        throwsA(
+          isA<GrpcError>().having(
+            (e) => e.codeName,
+            'Test codename',
+            contains('CANCELLED'),
+          ),
+        ),
+      );
+    },
+  );
+
+  test(
+    'Cancelling a call correctly complete trailers futures',
+    () async {
+      final clientCall = harness.client.unary(dummyValue);
+
+      Future.delayed(
+        Duration(milliseconds: cancelDurationMillis),
+      ).then((_) {
+        clientCall.cancel();
+      });
+
+      expect(
+        await clientCall.trailers,
+        isEmpty,
+      );
+
+      await expectLater(
+        clientCall,
+        throwsA(
+          isA<GrpcError>().having(
+            (e) => e.codeName,
+            'Test codename',
+            contains('CANCELLED'),
+          ),
+        ),
+      );
+    },
+  );
 }

test/keepalive_test.dart

@@ -49,7 +49,7 @@ void main() {
       services: [FakeEchoService()],
       keepAliveOptions: serverOptions,
     );
-    await server.serve(address: 'localhost', port: 8081);
+    await server.serve(address: 'localhost', port: 0);
     fakeChannel = FakeClientChannel(
       'localhost',
       port: server.port!,