fix(graphql): Consume streams from offline graphql requests (#617)

Author: elbieds

packages/brick_offline_first_with_graphql/lib/src/graphql_offline_request_queue.dart

@@ -22,6 +22,6 @@ class GraphqlOfflineRequestQueue extends OfflineRequestQueue<Request> {
   @override
   Future<void> transmitRequest(Request request) async {
     logger.finest('Processing request ${request.operation.operationName}');
-    link.request(request);
+    await link.request(request).drain();
   }
 }

packages/brick_offline_first_with_graphql/test/__helpers__.dart

@@ -1,5 +1,7 @@
 import 'dart:convert';
 
+import 'package:gql/ast.dart';
+import 'package:gql/language.dart';
 import 'package:gql_exec/gql_exec.dart';
 import 'package:gql_link/gql_link.dart';
 import 'package:mockito/mockito.dart';
@@ -55,3 +57,5 @@ class MockLink extends Mock implements Link {
         ),
       ) as Stream<Response>;
 }
+
+DocumentNode gql(String query) => parseString(query);

packages/brick_offline_first_with_graphql/test/graphql_offline_request_queue_test.dart

@@ -1,9 +1,15 @@
+import 'dart:async';
+
 import 'package:brick_offline_first_with_graphql/src/graphql_offline_queue_link.dart';
 import 'package:brick_offline_first_with_graphql/src/graphql_offline_request_queue.dart';
 import 'package:brick_offline_first_with_graphql/src/graphql_request_sqlite_cache_manager.dart';
+import 'package:gql_exec/gql_exec.dart';
+import 'package:mockito/mockito.dart';
 import 'package:sqflite_common_ffi/sqflite_ffi.dart';
 import 'package:test/test.dart';
 
+import '__helpers__.dart';
+
 void main() {
   final offlineClient = GraphqlOfflineQueueLink(
     GraphqlRequestSqliteCacheManager('db', databaseFactory: databaseFactoryFfi),
@@ -15,23 +21,85 @@ void main() {
       databaseFactory: databaseFactoryFfi,
     );
 
-    test('#start', () {
-      final queue = GraphqlOfflineRequestQueue(
-        link: offlineClient,
-        requestManager: requestManager,
-      )..start();
-      expect(queue.isRunning, isTrue);
-      queue.stop();
+    group('Queue Lifecycle', () {
+      test('#start', () {
+        final queue = GraphqlOfflineRequestQueue(
+          link: offlineClient,
+          requestManager: requestManager,
+        )..start();
+        expect(queue.isRunning, isTrue);
+        queue.stop();
+      });
+
+      test('#stop', () {
+        final queue = GraphqlOfflineRequestQueue(
+          link: offlineClient,
+          requestManager: requestManager,
+        )..start();
+        expect(queue.isRunning, isTrue);
+        queue.stop();
+        expect(queue.isRunning, isFalse);
+      });
     });
 
-    test('#stop', () {
-      final queue = GraphqlOfflineRequestQueue(
-        link: offlineClient,
-        requestManager: requestManager,
-      )..start();
-      expect(queue.isRunning, isTrue);
-      queue.stop();
-      expect(queue.isRunning, isFalse);
+    group('Request Processing', () {
+      test('#transmitRequest', () async {
+        final mockLink = MockLink();
+        var streamConsumed = false;
+
+        when(mockLink.request(any)).thenAnswer((_) {
+          return Stream<Response>.eventTransformed(
+            Stream.fromIterable([
+              const Response(
+                data: {'test': 'data'},
+                response: {'body': '{"test":"data"}'},
+              ),
+            ]),
+            (sink) {
+              streamConsumed = true;
+              return sink;
+            },
+          );
+        });
+
+        final testQueue = GraphqlOfflineRequestQueue(
+          link: mockLink,
+          requestManager: requestManager,
+        );
+
+        final testRequest = Request(
+          operation: Operation(
+            document: gql('mutation TestMutation { test }'),
+          ),
+        );
+
+        await testQueue.transmitRequest(testRequest);
+
+        expect(streamConsumed, isTrue);
+      });
+
+      test('#transmitRequest with errors', () async {
+        final mockLink = MockLink();
+        when(mockLink.request(any)).thenAnswer(
+          (_) => Stream.error(Exception('Network error')),
+        );
+
+        final testQueue = GraphqlOfflineRequestQueue(
+          link: mockLink,
+          requestManager: requestManager,
+        );
+
+        final testRequest = Request(
+          operation: Operation(
+            document: gql('mutation TestMutation { test }'),
+          ),
+        );
+
+        expect(
+          () => testQueue.transmitRequest(testRequest),
+          throwsA(isA<Exception>()),
+        );
+      });
     });
   });
 }