Close streams before database

drift/lib/src/runtime/executor/stream_queries.dart

@@ -178,7 +178,7 @@ class StreamQueryStore {
     _isShuttingDown = true;
 
     for (final stream in _activeKeyStreams.values) {
-      stream.close();
+      await stream.close();
     }
     // awaiting this is fine - the stream is never exposed to users and we don't
     // pause any subscriptions on it.
@@ -209,7 +209,7 @@ class QueryStream<Rows extends Object> {
       final queryListener = _QueryStreamListener(listener);
 
       if (_isClosed) {
-        listener.closeSync();
+        listener.close();
         return;
       }
 
@@ -358,12 +358,13 @@ class QueryStream<Rows extends Object> {
     }
   }
 
-  void close() {
+  Future<void> close() async {
     _isClosed = true;
-    for (final listener in _listeners) {
-      listener.controller.close();
-    }
+
+    final listenersDone = Future.wait(
+        [for (final listener in _listeners) listener.controller.close()]);
     _listeners.clear();
+    await listenersDone;
   }
 }
 

drift/test/database/streams_test.dart

@@ -234,6 +234,21 @@ void main() {
 
     subscription.resume();
     await subscription.cancel();
+  }, skip: 'testing out awaited streams');
+
+  test('closing database waits for streams', () async {
+    final stream = db.select(db.users).watch();
+    final subscription = stream.listen((_) {})..pause();
+
+    var closed = false;
+    db.close().then((_) => closed = true);
+    await pumpEventQueue();
+    expect(closed, isFalse);
+
+    subscription.resume();
+    await subscription.cancel();
+    await pumpEventQueue();
+    expect(closed, isTrue);
   });
 
   group('stream keys', () {

drift/test/integration_tests/drift_files_integration_test.dart

@@ -92,6 +92,7 @@ void main() {
 
     test('can be used in a query stream', () async {
       final stream = StreamQueue(db.readView().watch());
+      addTearDown(stream.cancel);
       const entry = Config(
         configKey: 'another_key',
         configValue: DriftAny('value'),

drift/test/integration_tests/regress_1599_test.dart

@@ -12,6 +12,7 @@ void main() {
     addTearDown(db.close);
 
     final query = StreamQueue(db.select(db.myView).watch());
+    addTearDown(query.cancel);
     await expectLater(query, emits(isEmpty));
 
     await db.into(db.config).insert(ConfigCompanion.insert(

drift/test/isolate_test.dart

@@ -243,6 +243,7 @@ void main() {
         );
       });
 
+      stream.cancel();
       await db.close();
     }