Fix stream cancelling for FetchPolicy.CacheAndNetwork, too

ferry/lib/src/client.dart

@@ -172,14 +172,30 @@ class Client {
   Stream<QueryResponse<T>> _networkResponseStream<T>(
       QueryRequest<T> queryRequest) {
     try {
-      return link.request(queryRequest).map((response) => QueryResponse(
-            queryRequest: queryRequest,
-            data: (response.data == null || response.data.isEmpty)
-                ? null
-                : queryRequest.parseData(response.data),
-            graphqlErrors: response.errors,
-            dataSource: DataSource.Link,
-          ));
+      return link.request(queryRequest).transform(
+            StreamTransformer.fromHandlers(
+              handleData: (response, sink) => sink.add(
+                QueryResponse(
+                  queryRequest: queryRequest,
+                  data: (response.data == null || response.data.isEmpty)
+                      ? null
+                      : queryRequest.parseData(response.data),
+                  graphqlErrors: response.errors,
+                  dataSource: DataSource.Link,
+                ),
+              ),
+              handleError: (error, stackTrace, sink) => sink.add(
+                QueryResponse<T>(
+                  queryRequest: queryRequest,
+                  linkException: error is LinkException
+                      ? error
+                      : ServerException(
+                          originalException: error, parsedResponse: null),
+                  dataSource: DataSource.Link,
+                ),
+              ),
+            ),
+          );
     } on LinkException catch (e) {
       return Stream.value(QueryResponse(
         queryRequest: queryRequest,

ferry/test/error_test.dart

@@ -68,4 +68,38 @@ void main() {
       expect(client.responseStream(allPokemonReq), emits(response));
     });
   });
+
+  group('Behavior after receiving errors', () {
+    test('Can emit data after emitting errors', () async {
+      final mockLink = MockLink();
+
+      final allPokemonReq = AllPokemon(
+          buildVars: (b) => b..first = 3,
+          fetchPolicy: FetchPolicy
+              .CacheAndNetwork // default is CacheFirst, which allows only 1 item from Link
+          );
+
+      when(mockLink.request(allPokemonReq, any)).thenAnswer((_) {
+        final controller = StreamController<Response>();
+
+        Future.delayed((Duration(microseconds: 1)))
+            .then((value) => controller.addError("error"));
+
+        Future.delayed((Duration(milliseconds: 100))).then((value) {
+          controller.add(Response(data: {}));
+          controller.close();
+        });
+
+        return controller.stream;
+      });
+
+      final client = Client(
+        link: mockLink,
+        options: ClientOptions(addTypename: false),
+      );
+
+      expect(client.responseStream(allPokemonReq),
+          emitsInOrder([isA<QueryResponse>(), isA<QueryResponse>()]));
+    });
+  });
 }